""" Service de gestion des badges """ from sqlalchemy.orm import Session from typing import List, Dict, Any from datetime import datetime from app.models.badge import Badge, UserBadge from app.models.user import User from app.models.user_prediction import UserPrediction from app.lib.badges import ( BADGES, isBadgeUnlocked, getBadgeById, ) class BadgeService: """Service pour la gestion des badges""" def __init__(self, db: Session): self.db = db def get_user_criteria(self, user_id: int) -> Dict[str, int]: """ Récupère les critères actuels de l'utilisateur """ # Nombre de prédictions consultées predictions_count = self.db.query(UserPrediction).filter( UserPrediction.user_id == user_id ).count() # Nombre de prédictions correctes (pour l'instant on utilise une valeur par défaut) # TODO: Implémenter la logique de calcul des prédictions correctes correct_predictions = 0 # Nombre de jours consécutifs (streak) # TODO: Implémenter la logique de calcul de streak streak_days = 0 # Nombre de partages # TODO: Implémenter la logique de suivi des partages share_count = 0 # Nombre de parrainages # TODO: Implémenter la logique de suivi des parrainages referral_count = 0 return { "predictions_count": predictions_count, "correct_predictions": correct_predictions, "streak_days": streak_days, "share_count": share_count, "referral_count": referral_count, } def check_and_unlock_badges(self, user_id: int) -> Dict[str, Any]: """ Vérifie et débloque les nouveaux badges pour un utilisateur """ # Récupérer les critères actuels de l'utilisateur user_criteria = self.get_user_criteria(user_id) # Récupérer les badges déjà débloqués par l'utilisateur unlocked_badges = self.db.query(UserBadge).filter( UserBadge.user_id == user_id ).all() unlocked_badge_ids = set() for ub in unlocked_badges: badge = self.db.query(Badge).filter(Badge.id == ub.badge_id).first() if badge: unlocked_badge_ids.add(badge.badge_id) # Vérifier tous les badges potentiels newly_unlocked_badges = [] for badge_def in BADGES: if badge_def["id"] not in unlocked_badge_ids: # Vérifier si le badge peut être débloqué criteria_type = badge_def["criteria"]["type"] criteria_value = badge_def["criteria"]["value"] if user_criteria[criteria_type] >= criteria_value: # Débloquer le badge new_badge = self._unlock_badge(user_id, badge_def) if new_badge: newly_unlocked_badges.append(new_badge) total_badges = len(unlocked_badges) + len(newly_unlocked_badges) # Générer un message de notification message = "" if len(newly_unlocked_badges) > 0: if len(newly_unlocked_badges) == 1: message = f'🎉 Félicitations ! Vous avez débloqué le badge "{newly_unlocked_badges[0]["name"]}" !' else: badge_names = ', '.join(b["name"] for b in newly_unlocked_badges) message = f'🎉 Félicitations ! Vous avez débloqué {len(newly_unlocked_badges)} nouveaux badges : {badge_names} !' return { "new_badges": newly_unlocked_badges, "total_badges": total_badges, "message": message, } def _unlock_badge(self, user_id: int, badge_def: Dict[str, Any]) -> Dict[str, Any] | None: """ Débloque un badge pour un utilisateur """ try: # Récupérer ou créer le badge dans la base de données db_badge = self.db.query(Badge).filter( Badge.badge_id == badge_def["id"] ).first() if not db_badge: # Créer le badge dans la base de données db_badge = Badge( badge_id=badge_def["id"], name=badge_def["name"], description=badge_def["description"], icon=badge_def["icon"], category=badge_def["category"], criteria_type=badge_def["criteria"]["type"], criteria_value=badge_def["criteria"]["value"], criteria_description=badge_def["criteria"]["description"], rarity=badge_def["rarity"], points=badge_def["points"], created_at=datetime.utcnow(), ) self.db.add(db_badge) self.db.flush() # Créer le badge utilisateur user_badge = UserBadge( user_id=user_id, badge_id=db_badge.id, unlocked_at=datetime.utcnow(), ) self.db.add(user_badge) self.db.commit() return { "id": db_badge.id, "badgeId": db_badge.badge_id, "name": db_badge.name, "description": db_badge.description, "icon": db_badge.icon, "category": db_badge.category, "rarity": db_badge.rarity, "points": db_badge.points, } except Exception as e: self.db.rollback() print(f"Erreur lors du déblocage du badge {badge_def['id']}: {e}") return None def get_user_badges(self, user_id: int) -> List[Dict[str, Any]]: """ Récupère tous les badges débloqués par un utilisateur """ user_badges = self.db.query(UserBadge).filter( UserBadge.user_id == user_id ).all() result = [] for ub in user_badges: badge = self.db.query(Badge).filter(Badge.id == ub.badge_id).first() if badge: result.append({ "id": ub.id, "userId": ub.user_id, "badgeId": ub.badge_id, "unlockedAt": ub.unlocked_at.isoformat(), "badge": { "id": badge.id, "badgeId": badge.badge_id, "name": badge.name, "description": badge.description, "icon": badge.icon, "category": badge.category, "rarity": badge.rarity, "points": badge.points, "createdAt": badge.created_at.isoformat(), } }) return result