81 lines
2.8 KiB
Python
81 lines
2.8 KiB
Python
from sqlmodel import Session, select, func, desc
|
|
from models import Performance, Show, Song, Attendance, UserBadge, Badge
|
|
from datetime import datetime
|
|
|
|
def get_song_stats(session: Session, song_id: int):
|
|
"""Calculate times played and gap for a song."""
|
|
# Times Played
|
|
times_played = session.exec(
|
|
select(func.count(Performance.id)).where(Performance.song_id == song_id)
|
|
).one()
|
|
|
|
# Last Played
|
|
last_performance = session.exec(
|
|
select(Show)
|
|
.join(Performance)
|
|
.where(Performance.song_id == song_id)
|
|
.order_by(desc(Show.date))
|
|
.limit(1)
|
|
).first()
|
|
|
|
gap = 0
|
|
if last_performance:
|
|
# Calculate gap: number of shows since last_performance
|
|
# This is a bit heavy if we count all shows.
|
|
# For now, let's just return the date of last played.
|
|
# To calculate true gap, we'd need: count(shows where date > last_performance.date)
|
|
gap = session.exec(
|
|
select(func.count(Show.id)).where(Show.date > last_performance.date)
|
|
).one()
|
|
|
|
# Set Breakdown
|
|
set_stats_query = session.exec(
|
|
select(Performance.set_name, func.count(Performance.id))
|
|
.where(Performance.song_id == song_id)
|
|
.group_by(Performance.set_name)
|
|
).all()
|
|
|
|
set_breakdown = {row[0]: row[1] for row in set_stats_query if row[0]}
|
|
|
|
return {
|
|
"times_played": times_played,
|
|
"last_played": last_performance.date if last_performance else None,
|
|
"gap": gap,
|
|
"set_breakdown": set_breakdown
|
|
}
|
|
|
|
def check_and_award_badges(session: Session, user_id: int):
|
|
"""Check for badge milestones and award them."""
|
|
# Example: 10 Shows Attended
|
|
attendance_count = session.exec(
|
|
select(func.count(Attendance.id)).where(Attendance.user_id == user_id)
|
|
).one()
|
|
|
|
if attendance_count >= 10:
|
|
award_badge(session, user_id, "10-shows", "10 Shows Club", "Awarded for attending 10 shows.", "ticket")
|
|
|
|
# Example: First Review
|
|
# (Need Review model import if implemented, assuming it is from previous steps)
|
|
# ...
|
|
|
|
def award_badge(session: Session, user_id: int, slug: str, name: str, description: str, icon: str):
|
|
"""Award a badge if not already owned."""
|
|
# Check if badge exists, create if not (for system badges)
|
|
badge = session.exec(select(Badge).where(Badge.slug == slug)).first()
|
|
if not badge:
|
|
badge = Badge(name=name, description=description, icon=icon, slug=slug)
|
|
session.add(badge)
|
|
session.commit()
|
|
session.refresh(badge)
|
|
|
|
# Check if user has it
|
|
user_badge = session.exec(
|
|
select(UserBadge)
|
|
.where(UserBadge.user_id == user_id)
|
|
.where(UserBadge.badge_id == badge.id)
|
|
).first()
|
|
|
|
if not user_badge:
|
|
user_badge = UserBadge(user_id=user_id, badge_id=badge.id)
|
|
session.add(user_badge)
|
|
session.commit()
|