ecoloop/api/services/benefit.py

from __future__ import annotations

from typing import Optional

from django.contrib.auth import get_user_model
from django.db.models import Sum, F, Max, Q, Window
from django.db.models.functions import Rank
from django.utils import timezone

from api.models import BenefitEvent


User = get_user_model()


# Default weights per action kind
BENEFIT_WEIGHTS = {
    BenefitEvent.KIND_ACCEPTED_ANSWER: 10,
    BenefitEvent.KIND_REVIEW_HELPFUL: 3,
    BenefitEvent.KIND_REFERRAL: 20,
    BenefitEvent.KIND_RECOMMENDATION: 5,
}


def log_benefit_event(
    *,
    benefactor: User,
    beneficiary: User,
    kind: str,
    points: Optional[int] = None,
    meta: Optional[dict] = None,
) -> BenefitEvent:
    if points is None:
        points = BENEFIT_WEIGHTS.get(kind, 1)
    event = BenefitEvent(
        benefactor=benefactor,
        beneficiary=beneficiary,
        kind=kind,
        points=points,
        meta=meta or {},
    )
    event.save()
    return event


def _period_start(period: str):
    period = (period or "").lower()
    if period in ("all", "all_time", "alltime", ""):
        return None
    now = timezone.now()
    if period in ("week", "weekly", "7d"):
        return now - timezone.timedelta(days=7)
    if period in ("month", "monthly", "30d"):
        return now - timezone.timedelta(days=30)
    # default fallback: weekly
    return now - timezone.timedelta(days=7)


def build_leaderboard_queryset(period: str = "weekly"):
    start = _period_start(period)
    pf = Q()
    if start is not None:
        pf = Q(benefit_given__created_at__gte=start)
    qs = (
        User.objects
        .annotate(
            points_given=Sum('benefit_given__points', filter=pf),
            last_help_at=Max('benefit_given__created_at', filter=pf),
        )
    )
    if start is not None:
        qs = qs.filter(benefit_given__created_at__gte=start)
    qs = (
        qs.filter(points_given__gt=0)
        .annotate(
            rank=Window(
                expression=Rank(),
                order_by=(
                    F('points_given').desc(nulls_last=True),
                    F('last_help_at').asc(nulls_last=True),
                    F('id').asc(),
                ),
            )
        )
        .order_by('rank')
    )
    return qs