""" 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) # awarded badges will be added to session but not committed here return awarded def get_leaderboard(session: Session, limit: int = 10) -> List[dict]: """Get top users by XP""" # Filter out test accounts # Filter out test accounts TEST_USER_EMAILS = [ "tenwest", "testuser", "rescue@elmeg.xyz", "admin-rescue@elmeg.xyz", "admin@elmeg.xyz", "test@", "emailtest" ] # Build query query = select(User).where(User.is_active == True) # Strictly filter out test accounts for test_email in TEST_USER_EMAILS: query = query.where(~User.email.ilike(f"%{test_email}%")) # Also honor the user's preference to hide from leaderboards # We use != False to include None (defaulting to True) or explicit True query = query.where(User.appear_in_leaderboards != False) # Hide users with 0 XP (inactive) query = query.where(User.xp > 0) users = session.exec( query.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") # ========================================== # USER TITLE & FLAIR SYSTEM (Tracker-style) # ========================================== # Titles that can be unlocked at certain levels LEVEL_TITLES = { 1: ["Rookie", "Newbie", "Fresh Ears"], 3: ["Fan", "Listener", "Devotee"], 5: ["Regular", "Familiar Face", "Couch Tour Pro"], 7: ["Veteran", "Road Warrior", "Tour Rat"], 10: ["Legend", "OG", "Scene Elder"], 12: ["Mythic", "Phenom", "Enlightened"], 15: ["Immortal", "Transcendent One", "Ascended"], } # Titles that can be purchased with XP PURCHASABLE_TITLES = { "Jam Connoisseur": {"cost": 500, "min_level": 3}, "Setlist Savant": {"cost": 750, "min_level": 5}, "Show Historian": {"cost": 1000, "min_level": 5}, "Type II Specialist": {"cost": 1500, "min_level": 7}, "Heady Scholar": {"cost": 2000, "min_level": 8}, "Rager": {"cost": 500, "min_level": 3}, "Rail Rider": {"cost": 750, "min_level": 4}, "Taper Section Regular": {"cost": 1000, "min_level": 5}, "Lot Lizard": {"cost": 600, "min_level": 4}, "Show Whisperer": {"cost": 2500, "min_level": 10}, } # Username colors that can be purchased with XP PURCHASABLE_COLORS = { "Sage Green": {"hex": "#6B9B6B", "cost": 300, "min_level": 2}, "Ocean Blue": {"hex": "#4A90D9", "cost": 300, "min_level": 2}, "Sunset Orange": {"hex": "#E67E22", "cost": 300, "min_level": 2}, "Royal Purple": {"hex": "#9B59B6", "cost": 500, "min_level": 4}, "Ruby Red": {"hex": "#E74C3C", "cost": 500, "min_level": 4}, "Electric Cyan": {"hex": "#00CED1", "cost": 750, "min_level": 6}, "Gold": {"hex": "#FFD700", "cost": 1000, "min_level": 8}, "Rainbow": {"hex": "gradient", "cost": 2500, "min_level": 10}, } # Flairs (small text/icon name beside username) PURCHASABLE_FLAIRS = { "Bolt": {"cost": 100, "min_level": 1}, "Guitar": {"cost": 100, "min_level": 1}, "Music": {"cost": 100, "min_level": 1}, "Rainbow": {"cost": 200, "min_level": 3}, "Fire": {"cost": 200, "min_level": 3}, "Star": {"cost": 300, "min_level": 5}, "Crown": {"cost": 500, "min_level": 7}, "Rocket": {"cost": 400, "min_level": 6}, "Diamond": {"cost": 750, "min_level": 9}, "Sparkles": {"cost": 1000, "min_level": 10}, } # Early adopter perks EARLY_ADOPTER_PERKS = { "free_title_change": True, # Early adopters can change title for free (once per month) "exclusive_titles": ["Pioneer", "Founding Member", "OG User", "Day One"], "exclusive_colors": {"Pioneer Gold": "#FFB347", "Genesis Green": "#50C878"}, "exclusive_flair": ["Medal", "Trophy"], "title_color": "#FFB347", # Default gold color for early adopters "bonus_xp_multiplier": 1.1, # 10% XP bonus } def get_available_titles(user: User) -> dict: """Get all titles available to this user based on level and status""" available = {} # Level-based titles for level, titles in LEVEL_TITLES.items(): if user.level >= level: for title in titles: available[title] = {"type": "level", "level_required": level, "cost": 0} # Purchasable titles for title, info in PURCHASABLE_TITLES.items(): if user.level >= info["min_level"]: available[title] = {"type": "purchase", "level_required": info["min_level"], "cost": info["cost"]} # Early adopter exclusive titles if user.is_early_adopter: for title in EARLY_ADOPTER_PERKS["exclusive_titles"]: available[title] = {"type": "early_adopter", "level_required": 1, "cost": 0} return available def get_available_colors(user: User) -> dict: """Get all colors available to this user""" available = {} for name, info in PURCHASABLE_COLORS.items(): if user.level >= info["min_level"]: available[name] = {"hex": info["hex"], "cost": info["cost"]} # Early adopter exclusive colors if user.is_early_adopter: for name, hex_color in EARLY_ADOPTER_PERKS["exclusive_colors"].items(): available[name] = {"hex": hex_color, "cost": 0} return available def get_available_flairs(user: User) -> dict: """Get all flairs available to this user""" available = {} for flair, info in PURCHASABLE_FLAIRS.items(): if user.level >= info["min_level"]: available[flair] = {"cost": info["cost"]} # Early adopter exclusive flairs if user.is_early_adopter: for flair in EARLY_ADOPTER_PERKS["exclusive_flair"]: available[flair] = {"cost": 0} return available def purchase_title(session: Session, user: User, title: str) -> Tuple[bool, str]: """Attempt to purchase a title. Returns (success, message)""" available = get_available_titles(user) if title not in available: return False, "Title not available at your level" info = available[title] cost = info["cost"] # Early adopters get free title changes for level/early_adopter titles if user.is_early_adopter and info["type"] in ["level", "early_adopter"]: cost = 0 if user.xp < cost: return False, f"Not enough XP. Need {cost}, have {user.xp}" # Deduct XP and set title user.xp -= cost user.custom_title = title session.add(user) return True, f"Title '{title}' acquired!" + (f" (-{cost} XP)" if cost > 0 else " (Free!)") def purchase_color(session: Session, user: User, color_name: str) -> Tuple[bool, str]: """Attempt to purchase a username color""" available = get_available_colors(user) if color_name not in available: return False, "Color not available at your level" info = available[color_name] cost = info["cost"] if user.xp < cost: return False, f"Not enough XP. Need {cost}, have {user.xp}" # Deduct XP and set color user.xp -= cost user.title_color = info["hex"] session.add(user) return True, f"Color '{color_name}' applied!" + (f" (-{cost} XP)" if cost > 0 else " (Free!)") def purchase_flair(session: Session, user: User, flair: str) -> Tuple[bool, str]: """Attempt to purchase a flair""" available = get_available_flairs(user) if flair not in available: return False, "Flair not available at your level" info = available[flair] cost = info["cost"] if user.xp < cost: return False, f"Not enough XP. Need {cost}, have {user.xp}" # Deduct XP and set flair user.xp -= cost user.flair = flair session.add(user) return True, f"Flair {flair} acquired!" + (f" (-{cost} XP)" if cost > 0 else " (Free!)") def get_user_display(user: User) -> dict: """Get the full display info for a user including title, color, flair""" username = user.email.split("@")[0] if user.email else "Anonymous" # Determine title to show display_title = user.custom_title if not display_title: display_title = LEVEL_NAMES.get(user.level, "User") return { "username": username, "title": display_title, "color": user.title_color, "flair": user.flair, "level": user.level, "xp": user.xp, "is_early_adopter": user.is_early_adopter, "is_supporter": user.is_supporter, }