""" Gamification Service - XP, Levels, Badges, and Streaks """ from datetime import datetime, timedelta from typing import Optional, List, Tuple from sqlmodel import Session, select, func from models import User, Badge, UserBadge, Attendance, Rating, Review, Comment # XP rewards for different actions XP_REWARDS = { "attendance_add": 25, # Mark a show as attended "rating_submit": 10, # Submit a rating "review_write": 50, # Write a review "comment_post": 5, # Post a comment "first_rating": 25, # First ever rating (bonus) "first_review": 50, # First ever review (bonus) "streak_bonus": 10, # Per day of streak "daily_login": 5, # Daily activity bonus } # Level thresholds (XP required for each level) LEVEL_THRESHOLDS = [ 0, # Level 1 100, # Level 2 250, # Level 3 500, # Level 4 1000, # Level 5 2000, # Level 6 3500, # Level 7 5500, # Level 8 8000, # Level 9 12000, # Level 10 18000, # Level 11 26000, # Level 12 36000, # Level 13 50000, # Level 14 70000, # Level 15 ] LEVEL_NAMES = { 1: "Rookie", 2: "Fan", 3: "Enthusiast", 4: "Regular", 5: "Dedicated", 6: "Veteran", 7: "Expert", 8: "Master", 9: "Elite", 10: "Legend", 11: "Icon", 12: "Mythic", 13: "Transcendent", 14: "Eternal", 15: "Immortal", } def calculate_level(xp: int) -> int: """Calculate level based on XP""" for level, threshold in enumerate(LEVEL_THRESHOLDS): if xp < threshold: return level # Previous level return len(LEVEL_THRESHOLDS) # Max level def xp_for_next_level(current_xp: int) -> Tuple[int, int]: """Returns (XP needed for next level, XP progress toward next level)""" current_level = calculate_level(current_xp) if current_level >= len(LEVEL_THRESHOLDS): return 0, 0 # Max level next_threshold = LEVEL_THRESHOLDS[current_level] prev_threshold = LEVEL_THRESHOLDS[current_level - 1] if current_level > 0 else 0 progress = current_xp - prev_threshold needed = next_threshold - prev_threshold return needed, progress def award_xp(session: Session, user: User, amount: int, reason: str) -> Tuple[int, bool]: """ Award XP to a user and check for level up. Returns (new_total_xp, did_level_up) """ old_level = user.level user.xp += amount new_level = calculate_level(user.xp) level_up = new_level > old_level if level_up: user.level = new_level session.add(user) return user.xp, level_up def update_streak(session: Session, user: User) -> int: """Update user's activity streak. Returns current streak.""" now = datetime.utcnow() if user.last_activity: days_since = (now.date() - user.last_activity.date()).days if days_since == 0: # Same day, no streak change pass elif days_since == 1: # Next day, increment streak user.streak_days += 1 # Award streak bonus award_xp(session, user, XP_REWARDS["streak_bonus"] * min(user.streak_days, 7), "streak_bonus") else: # Streak broken user.streak_days = 1 else: user.streak_days = 1 user.last_activity = now session.add(user) return user.streak_days # Badge definitions for seeding BADGE_DEFINITIONS = [ # Attendance badges {"name": "First Show", "slug": "first-show", "description": "Marked your first show as attended", "icon": "ticket", "tier": "bronze", "category": "attendance", "xp_reward": 50}, {"name": "Regular", "slug": "regular-10", "description": "Attended 10 shows", "icon": "calendar", "tier": "bronze", "category": "attendance", "xp_reward": 100}, {"name": "Veteran", "slug": "veteran-50", "description": "Attended 50 shows", "icon": "award", "tier": "silver", "category": "attendance", "xp_reward": 250}, {"name": "Lifer", "slug": "lifer-100", "description": "Attended 100 shows", "icon": "crown", "tier": "gold", "category": "attendance", "xp_reward": 500}, {"name": "Legend", "slug": "legend-250", "description": "Attended 250 shows", "icon": "star", "tier": "platinum", "category": "attendance", "xp_reward": 1000}, # Rating badges {"name": "First Rating", "slug": "first-rating", "description": "Submitted your first rating", "icon": "star", "tier": "bronze", "category": "ratings", "xp_reward": 25}, {"name": "Critic", "slug": "critic-50", "description": "Submitted 50 ratings", "icon": "thumbs-up", "tier": "silver", "category": "ratings", "xp_reward": 150}, {"name": "Connoisseur", "slug": "connoisseur-200", "description": "Submitted 200 ratings", "icon": "wine", "tier": "gold", "category": "ratings", "xp_reward": 400}, # Review badges {"name": "Wordsmith", "slug": "first-review", "description": "Wrote your first review", "icon": "pen", "tier": "bronze", "category": "social", "xp_reward": 50}, {"name": "Columnist", "slug": "columnist-10", "description": "Wrote 10 reviews", "icon": "file-text", "tier": "silver", "category": "social", "xp_reward": 200}, {"name": "Essayist", "slug": "essayist-50", "description": "Wrote 50 reviews", "icon": "book-open", "tier": "gold", "category": "social", "xp_reward": 500}, # Streak badges {"name": "Consistent", "slug": "streak-7", "description": "7-day activity streak", "icon": "flame", "tier": "bronze", "category": "milestones", "xp_reward": 75}, {"name": "Dedicated", "slug": "streak-30", "description": "30-day activity streak", "icon": "zap", "tier": "silver", "category": "milestones", "xp_reward": 300}, {"name": "Unstoppable", "slug": "streak-100", "description": "100-day activity streak", "icon": "rocket", "tier": "gold", "category": "milestones", "xp_reward": 750}, # Special badges {"name": "Debut Hunter", "slug": "debut-witness", "description": "Was in attendance for a song debut", "icon": "sparkles", "tier": "gold", "category": "milestones", "xp_reward": 200}, {"name": "Heady Spotter", "slug": "heady-witness", "description": "Attended a top-rated performance", "icon": "trophy", "tier": "silver", "category": "milestones", "xp_reward": 150}, {"name": "Song Chaser", "slug": "chase-caught-5", "description": "Caught 5 chase songs", "icon": "target", "tier": "silver", "category": "milestones", "xp_reward": 200}, ] def check_and_award_badges(session: Session, user: User) -> List[Badge]: """ Check all badge criteria and award any earned badges. Returns list of newly awarded badges. """ awarded = [] # Get user's existing badge slugs existing = session.exec( select(Badge.slug) .join(UserBadge) .where(UserBadge.user_id == user.id) ).all() existing_slugs = set(existing) # Count attendance attendance_count = session.exec( select(func.count(Attendance.id)) .where(Attendance.user_id == user.id) ).one() or 0 # Count ratings rating_count = session.exec( select(func.count(Rating.id)) .where(Rating.user_id == user.id) ).one() or 0 # Count reviews review_count = session.exec( select(func.count(Review.id)) .where(Review.user_id == user.id) ).one() or 0 badges_to_check = [ ("first-show", attendance_count >= 1), ("regular-10", attendance_count >= 10), ("veteran-50", attendance_count >= 50), ("lifer-100", attendance_count >= 100), ("legend-250", attendance_count >= 250), ("first-rating", rating_count >= 1), ("critic-50", rating_count >= 50), ("connoisseur-200", rating_count >= 200), ("first-review", review_count >= 1), ("columnist-10", review_count >= 10), ("essayist-50", review_count >= 50), ("streak-7", user.streak_days >= 7), ("streak-30", user.streak_days >= 30), ("streak-100", user.streak_days >= 100), ] for slug, condition in badges_to_check: if condition and slug not in existing_slugs: badge = session.exec(select(Badge).where(Badge.slug == slug)).first() if badge: user_badge = UserBadge(user_id=user.id, badge_id=badge.id) session.add(user_badge) award_xp(session, user, badge.xp_reward, f"badge_{slug}") awarded.append(badge) existing_slugs.add(slug) if awarded: session.commit() return awarded def get_leaderboard(session: Session, limit: int = 10) -> List[dict]: """Get top users by XP""" users = session.exec( select(User) .where(User.is_active == True) .order_by(User.xp.desc()) .limit(limit) ).all() return [ { "id": u.id, "email": u.email.split("@")[0], # Just username part "xp": u.xp, "level": u.level, "level_name": LEVEL_NAMES.get(u.level, "Unknown"), "streak": u.streak_days, } for u in users ] def seed_badges(session: Session): """Seed all badge definitions into the database""" for badge_def in BADGE_DEFINITIONS: existing = session.exec( select(Badge).where(Badge.slug == badge_def["slug"]) ).first() if not existing: badge = Badge(**badge_def) session.add(badge) session.commit() print(f"Seeded {len(BADGE_DEFINITIONS)} badge definitions")