scores.py

from sqlalchemy.sql.expression import union_all
from sqlalchemy import func

from CTFd.cache import cache
from CTFd.models import db, Teams, Users, Solves, Awards, Challenges, TeamFieldEntries, Fields
from CTFd.utils.dates import unix_time_to_utc
from CTFd.utils import get_config
from CTFd.utils.modes import get_model

@cache.memoize(timeout=60)
def get_team_ids():
    attr_id = get_config("split_scoreboard_attr", 0)
    attr_value = get_config("split_scoreboard_value", "hidden")

    team_ids = []
    if attr_id == "-1": # Where team size is <value>
        teams = Teams.query.outerjoin(Teams.members).group_by(Teams).having(func.count_(Teams.members) == attr_value)
        for team in teams:
            team_ids.append(team.id)
    elif attr_id == "-2": # Where team size is less than <value> 
        teams = Teams.query.outerjoin(Teams.members).group_by(Teams).having(func.count_(Teams.members) <= attr_value)
        for team in teams:
            team_ids.append(team.id)
    elif attr_id == "-3": # Where team size is greater than <value> 
        teams = Teams.query.outerjoin(Teams.members).group_by(Teams).having(func.count_(Teams.members) >= attr_value)
        for team in teams:
            team_ids.append(team.id)
    else:
        teams = TeamFieldEntries.query.filter_by(
            field_id = attr_id
        ).filter(
            func.lower(TeamFieldEntries.value) == func.lower(str(attr_value))
        )
        for team in teams:
            team_ids.append(team.team_id)

    return team_ids


#@cache.memoize(timeout=60)
def get_scores(admin=False):
    scores = (
        db.session.query(
            Solves.account_id.label("account_id"),
            db.func.sum(Challenges.value).label("score"),
            db.func.max(Solves.id).label("id"),
            db.func.max(Solves.date).label("date"),
        )
        .join(Challenges)
        .filter(Challenges.value != 0)
        .group_by(Solves.account_id)
    )

    awards = (
        db.session.query(
            Awards.account_id.label("account_id"),
            db.func.sum(Awards.value).label("score"),
            db.func.max(Awards.id).label("id"),
            db.func.max(Awards.date).label("date"),
        )
        .filter(Awards.value != 0)
        .group_by(Awards.account_id)
    )

    """
    Filter out solves and awards that are before a specific time point.
    """
    freeze = get_config("freeze")
    if not admin and freeze:
        scores = scores.filter(Solves.date < unix_time_to_utc(freeze))
        awards = awards.filter(Awards.date < unix_time_to_utc(freeze))

    """
    Combine awards and solves with a union. They should have the same amount of columns
    """
    results = union_all(scores, awards).alias("results")

    """
    Sum each of the results by the team id to get their score.
    """
    sumscores = (
        db.session.query(
            results.columns.account_id,
            db.func.sum(results.columns.score).label("score"),
            db.func.max(results.columns.id).label("id"),
            db.func.max(results.columns.date).label("date"),
        )
        .group_by(results.columns.account_id)
        .subquery()
    )

    return sumscores

@cache.memoize(timeout=60)
def get_unmatched_standings(count=None, admin=False, fields=[]):
    """
    Get standings as a list of tuples containing account_id, name, and score e.g. [(account_id, team_name, score)].
    Ties are broken by who reached a given score first based on the solve ID. Two users can have the same score but one
    user will have a solve ID that is before the others. That user will be considered the tie-winner.
    Challenges & Awards with a value of zero are filtered out of the calculations to avoid incorrect tie breaks.
    """

    if fields is None:
        fields = []
    Model = get_model()
    team_ids = get_team_ids()
    sumscores = get_scores(admin)

    """
    Admins can see scores for all users but the public cannot see banned users.
    Filters out banned users.
    Properly resolves value ties by ID.
    Different databases treat time precision differently so resolve by the row ID instead.
    """
    if admin:
        standings_query = (
            db.session.query(
                Model.id.label("account_id"),
                Model.oauth_id.label("oauth_id"),
                Model.name.label("name"),
                Model.hidden,
                Model.banned,
                sumscores.columns.score,
                *fields,
            )
            .join(sumscores, Model.id == sumscores.columns.account_id)
			.filter(Model.id.notin_(team_ids))
            .order_by(sumscores.columns.score.desc(), sumscores.columns.id)
        )
    else:
        standings_query = (
            db.session.query(
                Model.id.label("account_id"),
                Model.oauth_id.label("oauth_id"),
                Model.name.label("name"),
                sumscores.columns.score,
                *fields,
            )
            .join(sumscores, Model.id == sumscores.columns.account_id)
            .filter(Model.banned == False, Model.hidden == False)
			.filter(Model.id.notin_(team_ids))
            .order_by(sumscores.columns.score.desc(), sumscores.columns.id)
        )

    """
    Only select a certain amount of users if asked.
    """
    if count is None:
        standings = standings_query.all()
    else:
        standings = standings_query.limit(count).all()

    return standings

@cache.memoize(timeout=60)
def get_custom_standings(count=None, admin=False, team_ids=[], fields=[]):

    if fields is None:
        fields = []
    Model = get_model()
    sumscores = get_scores(admin)

    """
    Admins can see scores for all users but the public cannot see banned users.
    Filters out banned users.
    Properly resolves value ties by ID.
    Different databases treat time precision differently so resolve by the row ID instead.
    """
    if admin:
        standings_query = (
            db.session.query(
                Model.id.label("account_id"),
                Model.oauth_id.label("oauth_id"),
                Model.name.label("name"),
                Model.hidden,
                Model.banned,
                sumscores.columns.score,
                *fields,
            )
            .join(sumscores, Model.id == sumscores.columns.account_id)
			.filter(Model.id.in_(team_ids))
            .order_by(sumscores.columns.score.desc(), sumscores.columns.id)
        )
    else:
        standings_query = (
            db.session.query(
                Model.id.label("account_id"),
                Model.oauth_id.label("oauth_id"),
                Model.name.label("name"),
                sumscores.columns.score,
                *fields,
            )
            .join(sumscores, Model.id == sumscores.columns.account_id)
            .filter(Model.banned == False, Model.hidden == False)
			.filter(Model.id.in_(team_ids))
            .order_by(sumscores.columns.score.desc(), sumscores.columns.id)
        )

    """
    Only select a certain amount of users if asked.
    """
    if count is None:
        standings = standings_query.all()
    else:
        standings = standings_query.limit(count).all()

    return standings


@cache.memoize(timeout=60)
def get_matched_standings(count=None, admin=False, fields=None):
    
    if fields is None:
        fields = []
    Model = get_model()
    team_ids = get_team_ids()
    sumscores = get_scores(admin)

    """
    Admins can see scores for all users but the public cannot see banned users.
    Filters out banned users.
    Properly resolves value ties by ID.
    Different databases treat time precision differently so resolve by the row ID instead.
    """
    if admin:
        standings_query = (
            db.session.query(
                Model.id.label("account_id"),
                Model.oauth_id.label("oauth_id"),
                Model.name.label("name"),
                Model.hidden,
                Model.banned,
                sumscores.columns.score,
                *fields,
            )
            .join(sumscores, Model.id == sumscores.columns.account_id)
			.filter(Model.id.in_(team_ids))
            .order_by(sumscores.columns.score.desc(), sumscores.columns.id)
        )
    else:
        standings_query = (
            db.session.query(
                Model.id.label("account_id"),
                Model.oauth_id.label("oauth_id"),
                Model.name.label("name"),
                sumscores.columns.score,
                *fields,
            )
            .join(sumscores, Model.id == sumscores.columns.account_id)
            .filter(Model.banned == False, Model.hidden == False)
			.filter(Model.id.in_(team_ids))
            .order_by(sumscores.columns.score.desc(), sumscores.columns.id)
        )

    """
    Only select a certain amount of users if asked.
    """
    if count is None:
        standings = standings_query.all()
    else:
        standings = standings_query.limit(count).all()

    return standings