fediversion/backend/services/gamification.py
fullsizemalt b4cddf41ea feat: Initialize Fediversion multi-band platform
- Fork elmeg-demo codebase for multi-band support
- Add data importer infrastructure with base class
- Create band-specific importers:
  - phish.py: Phish.net API v5
  - grateful_dead.py: Grateful Stats API
  - setlistfm.py: Dead & Company, Billy Strings (Setlist.fm)
- Add spec-kit configuration for Gemini
- Update README with supported bands and architecture
2025-12-28 12:39:28 -08:00

488 lines
17 KiB
Python

"""
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,
}