"""Leaderboard service with examples""" from datetime import datetime, timedelta, timezone from sqlalchemy import func, desc from app.extensions import db from app.models.user import User from app.models.gamification import PointActivity import logging logger = logging.getLogger(__name__) class LeaderboardService: @staticmethod def get_weekly_leaderboard(limit=10): """ Get this week's top users. Example response: { 'leaderboard': [ {'rank': 1, 'user_id': 1, 'name': 'John Doe', 'points': 500}, {'rank': 2, 'user_id': 2, 'name': 'Jane Smith', 'points': 450} ], 'period': {'start': '2025-01-13', 'end': '2025-01-19'} } """ try: today = datetime.now(timezone.utc).date() week_start = today - timedelta(days=today.weekday()) week_end = week_start + timedelta(days=6) results = db.session.query( PointActivity.user_id, func.sum(PointActivity.points_earned).label('points') ).filter( PointActivity.created_at >= week_start, PointActivity.created_at <= week_end + timedelta(days=1) # Include full end day ).group_by( PointActivity.user_id ).order_by( desc('points') ).limit(limit).all() leaderboard = [] for rank, (user_id, points) in enumerate(results, 1): user = User.query.get(user_id) if user: leaderboard.append({ 'rank': rank, 'user_id': user_id, 'name': f"{user.first_name} {user.last_name}", 'role': user.role, 'points': int(points), 'level': user.current_level or 1, 'league': user.seasonal_league or 'bronze' }) return { 'leaderboard': leaderboard, 'period': { 'start': week_start.isoformat(), 'end': week_end.isoformat(), 'type': 'weekly' } } except Exception as e: logger.error(f"Error getting weekly leaderboard: {e}") return { 'leaderboard': [], 'period': {'start': None, 'end': None, 'type': 'weekly'}, 'error': 'Failed to fetch leaderboard' } @staticmethod def get_monthly_leaderboard(limit=20): """Get this month's top users""" try: today = datetime.now(timezone.utc).date() month_start = today.replace(day=1) # Calculate last day of month if today.month == 12: month_end = today.replace(year=today.year + 1, month=1, day=1) - timedelta(days=1) else: month_end = today.replace(month=today.month + 1, day=1) - timedelta(days=1) results = db.session.query( PointActivity.user_id, func.sum(PointActivity.points_earned).label('points') ).filter( PointActivity.created_at >= month_start, PointActivity.created_at <= month_end + timedelta(days=1) ).group_by( PointActivity.user_id ).order_by( desc('points') ).limit(limit).all() leaderboard = [] for rank, (user_id, points) in enumerate(results, 1): user = User.query.get(user_id) if user: leaderboard.append({ 'rank': rank, 'user_id': user_id, 'name': f"{user.first_name} {user.last_name}", 'role': user.role, 'points': int(points), 'level': user.current_level or 1, 'league': user.seasonal_league or 'bronze' }) return { 'leaderboard': leaderboard, 'period': { 'start': month_start.isoformat(), 'end': month_end.isoformat(), 'type': 'monthly' } } except Exception as e: logger.error(f"Error getting monthly leaderboard: {e}") return { 'leaderboard': [], 'period': {'start': None, 'end': None, 'type': 'monthly'}, 'error': 'Failed to fetch leaderboard' } @staticmethod def get_all_time_leaderboard(limit=50): """Get all-time top users based on total points""" try: results = User.query.filter( User.total_points > 0 ).order_by( desc(User.total_points) ).limit(limit).all() leaderboard = [] for rank, user in enumerate(results, 1): leaderboard.append({ 'rank': rank, 'user_id': user.id, 'name': f"{user.first_name} {user.last_name}", 'role': user.role, 'points': user.total_points or 0, 'level': user.current_level or 1, 'league': user.seasonal_league or 'bronze', 'jobs_completed': user.jobs_completed or 0, 'average_rating': float(user.average_rating) if user.average_rating else 0.0 }) return { 'leaderboard': leaderboard, 'period': { 'type': 'all_time' } } except Exception as e: logger.error(f"Error getting all-time leaderboard: {e}") return { 'leaderboard': [], 'period': {'type': 'all_time'}, 'error': 'Failed to fetch leaderboard' } @staticmethod def get_league_leaderboard(league: str, limit=25): """Get leaderboard for specific league (bronze, silver, gold, platinum, diamond)""" try: if league not in ['bronze', 'silver', 'gold', 'platinum', 'diamond']: league = 'bronze' results = User.query.filter( User.seasonal_league == league, User.total_points > 0 ).order_by( desc(User.total_points) ).limit(limit).all() leaderboard = [] for rank, user in enumerate(results, 1): leaderboard.append({ 'rank': rank, 'user_id': user.id, 'name': f"{user.first_name} {user.last_name}", 'role': user.role, 'points': user.total_points or 0, 'level': user.current_level or 1, 'league': user.seasonal_league or 'bronze', 'jobs_completed': user.jobs_completed or 0, 'average_rating': float(user.average_rating) if user.average_rating else 0.0 }) return { 'leaderboard': leaderboard, 'league': league, 'total_in_league': len(results) } except Exception as e: logger.error(f"Error getting {league} league leaderboard: {e}") return { 'leaderboard': [], 'league': league, 'total_in_league': 0, 'error': 'Failed to fetch league leaderboard' } @staticmethod def get_user_position(user_id: int, leaderboard_type: str = 'all_time'): """Get specific user's position in leaderboard""" try: user = User.query.get(user_id) if not user: return {'position': 0, 'total_users': 0, 'user_points': 0} if leaderboard_type == 'weekly': today = datetime.now(timezone.utc).date() week_start = today - timedelta(days=today.weekday()) week_end = week_start + timedelta(days=6) # Get user's weekly points user_weekly_points = db.session.query( func.sum(PointActivity.points_earned) ).filter( PointActivity.user_id == user_id, PointActivity.created_at >= week_start, PointActivity.created_at <= week_end + timedelta(days=1) ).scalar() or 0 # Count users with more weekly points better_users = db.session.query( PointActivity.user_id, func.sum(PointActivity.points_earned).label('points') ).filter( PointActivity.created_at >= week_start, PointActivity.created_at <= week_end + timedelta(days=1) ).group_by( PointActivity.user_id ).having( func.sum(PointActivity.points_earned) > user_weekly_points ).count() total_weekly_users = db.session.query( PointActivity.user_id ).filter( PointActivity.created_at >= week_start, PointActivity.created_at <= week_end + timedelta(days=1) ).distinct().count() return { 'position': better_users + 1, 'total_users': total_weekly_users, 'user_points': int(user_weekly_points), 'type': 'weekly' } else: # all_time better_users = User.query.filter( User.total_points > (user.total_points or 0) ).count() total_users = User.query.filter( User.total_points > 0 ).count() return { 'position': better_users + 1, 'total_users': total_users, 'user_points': user.total_points or 0, 'type': 'all_time' } except Exception as e: logger.error(f"Error getting user position: {e}") return {'position': 0, 'total_users': 0, 'user_points': 0, 'error': str(e)} @staticmethod def get_league_promotion_candidates(): """Get users who might be ready for league promotion""" try: # Simple promotion logic based on points and performance promotion_candidates = [] # Bronze to Silver: 500+ points, 3+ jobs, 4.0+ rating bronze_users = User.query.filter( User.seasonal_league == 'bronze', User.total_points >= 500, User.jobs_completed >= 3, User.average_rating >= 4.0 ).all() for user in bronze_users: promotion_candidates.append({ 'user_id': user.id, 'current_league': 'bronze', 'suggested_league': 'silver', 'points': user.total_points, 'jobs_completed': user.jobs_completed, 'average_rating': float(user.average_rating) if user.average_rating else 0.0 }) # Silver to Gold: 1500+ points, 10+ jobs, 4.5+ rating silver_users = User.query.filter( User.seasonal_league == 'silver', User.total_points >= 1500, User.jobs_completed >= 10, User.average_rating >= 4.5 ).all() for user in silver_users: promotion_candidates.append({ 'user_id': user.id, 'current_league': 'silver', 'suggested_league': 'gold', 'points': user.total_points, 'jobs_completed': user.jobs_completed, 'average_rating': float(user.average_rating) if user.average_rating else 0.0 }) return { 'candidates': promotion_candidates, 'total_candidates': len(promotion_candidates) } except Exception as e: logger.error(f"Error getting promotion candidates: {e}") return {'candidates': [], 'total_candidates': 0, 'error': str(e)} # Factory function for Flask app context def get_leaderboard_service(): """Factory to get leaderboard service""" return LeaderboardService()