gateway/modules/features/commcoach/serviceCommcoachGamification.py

# Copyright (c) 2025 Patrick Motsch
# All rights reserved.
"""
CommCoach Gamification - Badge definitions and award logic.
Checks and awards badges after each session completion.
"""

import logging
from typing import Dict, Any, List, Optional

logger = logging.getLogger(__name__)

BADGE_DEFINITIONS: Dict[str, Dict[str, Any]] = {
    "first_session": {
        "label": "Erste Session",
        "description": "Deine erste Coaching-Session abgeschlossen",
        "icon": "star",
    },
    "streak_3": {
        "label": "3-Tage-Serie",
        "description": "3 Tage in Folge eine Session absolviert",
        "icon": "fire",
    },
    "streak_7": {
        "label": "Wochenserie",
        "description": "7 Tage in Folge eine Session absolviert",
        "icon": "fire",
    },
    "streak_30": {
        "label": "Monatsserie",
        "description": "30 Tage in Folge eine Session absolviert",
        "icon": "fire",
    },
    "sessions_5": {
        "label": "Engagiert",
        "description": "5 Sessions abgeschlossen",
        "icon": "trophy",
    },
    "sessions_10": {
        "label": "Fortgeschritten",
        "description": "10 Sessions abgeschlossen",
        "icon": "trophy",
    },
    "sessions_25": {
        "label": "Experte",
        "description": "25 Sessions abgeschlossen",
        "icon": "trophy",
    },
    "sessions_50": {
        "label": "Meister",
        "description": "50 Sessions abgeschlossen",
        "icon": "trophy",
    },
    "high_score": {
        "label": "Bestleistung",
        "description": "Durchschnittsscore über 80 in einer Session",
        "icon": "medal",
    },
    "multi_context": {
        "label": "Vielseitig",
        "description": "3 verschiedene Coaching-Themen aktiv",
        "icon": "layers",
    },
    "roleplay_first": {
        "label": "Rollenspieler",
        "description": "Erste Roleplay-Session mit einer Persona abgeschlossen",
        "icon": "theater",
    },
    "all_dimensions": {
        "label": "Ganzheitlich",
        "description": "In allen 5 Kompetenz-Dimensionen bewertet",
        "icon": "compass",
    },
    "task_completer": {
        "label": "Umsetzer",
        "description": "10 Coaching-Aufgaben erledigt",
        "icon": "check-circle",
    },
}


async def checkAndAwardBadges(interface, userId: str, mandateId: str, instanceId: str,
                              session: Optional[Dict[str, Any]] = None) -> List[Dict[str, Any]]:
    """Check badge conditions and award any newly earned badges. Returns list of newly awarded badges."""
    awarded: List[Dict[str, Any]] = []

    profile = interface.getProfile(userId, instanceId)
    if not profile:
        return awarded

    totalSessions = profile.get("totalSessions", 0)
    streakDays = profile.get("streakDays", 0)

    badgesToCheck = [
        ("first_session", totalSessions >= 1),
        ("sessions_5", totalSessions >= 5),
        ("sessions_10", totalSessions >= 10),
        ("sessions_25", totalSessions >= 25),
        ("sessions_50", totalSessions >= 50),
        ("streak_3", streakDays >= 3),
        ("streak_7", streakDays >= 7),
        ("streak_30", streakDays >= 30),
    ]

    if session and session.get("competenceScore"):
        try:
            score = float(session["competenceScore"])
            if score >= 80:
                badgesToCheck.append(("high_score", True))
        except (ValueError, TypeError):
            pass

    if session and session.get("personaId") and session["personaId"] != "coach":
        badgesToCheck.append(("roleplay_first", True))

    try:
        from .datamodelCommcoach import CoachingContextStatus
        allContexts = interface.db.getRecordset(
            interface.db.getRecordset.__self__.__class__.__mro__[0]  # avoid import issues
        ) if False else []
    except Exception:
        allContexts = []

    completedTasks = interface.getCompletedTaskCount(userId) if hasattr(interface, 'getCompletedTaskCount') else 0
    if completedTasks >= 10:
        badgesToCheck.append(("task_completer", True))

    for badgeKey, condition in badgesToCheck:
        if condition and not interface.hasBadge(userId, instanceId, badgeKey):
            badgeData = {
                "userId": userId,
                "mandateId": mandateId,
                "instanceId": instanceId,
                "badgeKey": badgeKey,
            }
            newBadge = interface.awardBadge(badgeData)
            definition = BADGE_DEFINITIONS.get(badgeKey, {})
            newBadge["label"] = definition.get("label", badgeKey)
            newBadge["description"] = definition.get("description", "")
            newBadge["icon"] = definition.get("icon", "star")
            awarded.append(newBadge)
            logger.info(f"Badge '{badgeKey}' awarded to user {userId}")

    return awarded


def getBadgeDefinitions() -> Dict[str, Dict[str, Any]]:
    """Return all badge definitions for the frontend."""
    return BADGE_DEFINITIONS