54 lines
1.6 KiB
Python
54 lines
1.6 KiB
Python
from __future__ import annotations
|
|
|
|
from collections import defaultdict
|
|
from datetime import date, timedelta
|
|
|
|
from ..models import BadgeDefinition, TaskInstance
|
|
|
|
|
|
def _max_day_streak(days: set[date]) -> int:
|
|
if not days:
|
|
return 0
|
|
streak = 1
|
|
best = 1
|
|
ordered = sorted(days)
|
|
for previous, current in zip(ordered, ordered[1:]):
|
|
if current == previous + timedelta(days=1):
|
|
streak += 1
|
|
else:
|
|
streak = 1
|
|
best = max(best, streak)
|
|
return best
|
|
|
|
|
|
def compute_badge_awards(definitions: list[BadgeDefinition], completed_tasks: list[TaskInstance]) -> list[dict]:
|
|
by_type: dict[str, int] = defaultdict(int)
|
|
completion_days: set[date] = set()
|
|
|
|
for task in completed_tasks:
|
|
if not task.completed_at:
|
|
continue
|
|
completion_day = task.completed_at.date()
|
|
completion_days.add(completion_day)
|
|
by_type["monthly_task_count"] += 1
|
|
if task.due_date and completion_day < task.due_date:
|
|
by_type["early_finisher_count"] += 1
|
|
if task.due_date and completion_day <= task.due_date:
|
|
by_type["on_time_count"] += 1
|
|
|
|
by_type["streak_days"] = _max_day_streak(completion_days)
|
|
|
|
awards = []
|
|
for definition in definitions:
|
|
metric_value = by_type.get(definition.trigger_type, 0)
|
|
if definition.active and metric_value >= definition.threshold:
|
|
awards.append(
|
|
{
|
|
"definition": definition,
|
|
"metric_value": metric_value,
|
|
"bonus_points": definition.bonus_points,
|
|
}
|
|
)
|
|
return awards
|
|
|