174 lines
5.7 KiB
Python
174 lines
5.7 KiB
Python
"""Helper functions for team views and actions"""
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import func, inspect
|
|
from datetime import datetime, timedelta
|
|
import random
|
|
|
|
from ...models import Team, TeamMembership, User, QRCode
|
|
|
|
def get_team_members_with_details(db: Session, team_id: int):
|
|
"""Get team members with additional details"""
|
|
memberships = db.query(TeamMembership).filter_by(team_id=team_id).all()
|
|
team_members = []
|
|
|
|
for membership in memberships:
|
|
member = db.query(User).filter_by(id=membership.user_id).first()
|
|
if member:
|
|
# Use joined_at if available, otherwise use placeholder
|
|
joined_date = membership.joined_at or datetime.now() - timedelta(days=random.randint(30, 180))
|
|
if isinstance(joined_date, datetime):
|
|
month_name = joined_date.strftime("%b")
|
|
year = joined_date.strftime("%Y")
|
|
else:
|
|
month_name = "Apr"
|
|
year = "2023"
|
|
|
|
team_members.append({
|
|
"user": member,
|
|
"is_admin": membership.is_admin,
|
|
"is_captain": membership.is_captain,
|
|
"joined": f"{month_name} {year}"
|
|
})
|
|
|
|
return team_members
|
|
|
|
def check_user_permissions(db: Session, user, team_id: int, team):
|
|
"""Check if user is admin or owner of the team"""
|
|
is_user_admin = False
|
|
is_user_owner = False
|
|
is_captain = False
|
|
|
|
if user:
|
|
# Check admin status
|
|
membership = db.query(TeamMembership).filter_by(
|
|
team_id=team_id,
|
|
user_id=user.id,
|
|
is_admin=True
|
|
).first()
|
|
is_user_admin = membership is not None
|
|
|
|
# Check captain status
|
|
captain_membership = db.query(TeamMembership).filter_by(
|
|
team_id=team_id,
|
|
user_id=user.id,
|
|
is_captain=True
|
|
).first()
|
|
is_captain = captain_membership is not None
|
|
|
|
# Check owner status
|
|
is_user_owner = hasattr(team, 'owner_id') and team.owner_id == user.id
|
|
if is_user_owner:
|
|
is_user_admin = True # Owner has admin privileges
|
|
|
|
return is_user_admin, is_user_owner, is_captain
|
|
|
|
def get_team_total_points(db: Session, team_id: int):
|
|
"""Get total points for a team"""
|
|
total_points = db.query(func.sum(QRCode.points)).filter(
|
|
QRCode.redeemed_at_team == team_id
|
|
).scalar() or 0
|
|
|
|
return total_points
|
|
|
|
def calculate_team_rank(db: Session, team_id: int):
|
|
"""Calculate team rank based on points"""
|
|
try:
|
|
team = db.query(Team).filter(Team.id == team_id).first()
|
|
if not team:
|
|
return 1
|
|
|
|
# First, get the aggregated points for teams in the same league
|
|
team_points = db.query(
|
|
Team.id.label("team_id"),
|
|
func.sum(QRCode.points).label('total')
|
|
).outerjoin(
|
|
QRCode,
|
|
QRCode.redeemed_at_team == Team.id
|
|
).filter(
|
|
Team.league_id == team.league_id
|
|
).group_by(Team.id).all()
|
|
|
|
# Sort them by points (descending)
|
|
sorted_teams = sorted(team_points, key=lambda x: x.total or 0, reverse=True)
|
|
|
|
# Find our team's position
|
|
team_rank = 1
|
|
for idx, team_data in enumerate(sorted_teams):
|
|
if team_data.team_id == team_id:
|
|
team_rank = idx + 1
|
|
break
|
|
|
|
return team_rank
|
|
except Exception as e:
|
|
print(f"Error calculating team rank: {e}")
|
|
return 1 # Default to 1st place on error
|
|
|
|
def get_team_points_history(db: Session, team_id: int):
|
|
"""Get points history for a team"""
|
|
# Default values
|
|
points_this_month = 65
|
|
point_change = 15
|
|
point_change_positive = True
|
|
|
|
# Try to get actual data if available
|
|
try:
|
|
now = datetime.now()
|
|
first_day_of_month = datetime(now.year, now.month, 1)
|
|
|
|
# Use raw SQL to check if column exists and get points
|
|
has_redeemed_at = False
|
|
inspector = inspect(db.bind)
|
|
if 'redeemed_at' in [col['name'] for col in inspector.get_columns('qr_codes')]:
|
|
has_redeemed_at = True
|
|
|
|
if has_redeemed_at:
|
|
points_this_month = db.query(func.sum(QRCode.points)).filter(
|
|
QRCode.redeemed_at_team == team_id,
|
|
QRCode.redeemed_at >= first_day_of_month
|
|
).scalar() or points_this_month
|
|
except Exception as e:
|
|
print(f"Error calculating monthly points: {e}")
|
|
|
|
return points_this_month, point_change, point_change_positive
|
|
|
|
def get_team_activities():
|
|
"""Get team activities (currently returns mock data)"""
|
|
return [
|
|
{
|
|
"type": "points",
|
|
"points": 15,
|
|
"event": "Music Trivia Night",
|
|
"date": "September 12, 2023"
|
|
},
|
|
{
|
|
"type": "join",
|
|
"user": "Robert Brown",
|
|
"date": "July 28, 2023"
|
|
},
|
|
{
|
|
"type": "achievement",
|
|
"achievement": "1st place",
|
|
"event": "History Night",
|
|
"date": "July 15, 2023"
|
|
},
|
|
{
|
|
"type": "points",
|
|
"points": 20,
|
|
"event": "Movie Trivia Night",
|
|
"date": "July 1, 2023"
|
|
}
|
|
]
|
|
|
|
def get_team_age(team):
|
|
"""Calculate team age"""
|
|
created_at = getattr(team, 'created_at', None)
|
|
|
|
if created_at and isinstance(created_at, datetime):
|
|
days_ago = (datetime.now() - created_at).days
|
|
founded_date_str = created_at.strftime("%B %d, %Y")
|
|
else:
|
|
days_ago = 164 # Default fallback
|
|
founded_date_str = "March 22, 2023" # Default fallback
|
|
|
|
return days_ago, founded_date_str
|