org_score.py

import decimal
from flask import g, render_template, request
from flask.helpers import flash, url_for
from typing import Iterable, List, Optional, Tuple, Union
from flask_wtf.form import FlaskForm
import json
import werkzeug.exceptions
from werkzeug.utils import redirect
import wtforms

import mo
import mo.db as db
from mo.rights import Right
from mo.score import Score
from mo.web import app
from mo.web.org_contest import get_context
from mo.web.table import Cell, CellInput, CellLink, Column, Row, Table, OrderCell, cell_pion_link
from mo.util_format import format_decimal, inflect_number
from mo.web.user import scoretable_construct


class SolPointsCell(Cell):
    contest_id: int
    user: db.User
    sol: Optional[db.Solution]
    link_to_paper: bool

    def __init__(self, contest_id: int, user: db.User, sol: Optional[db.Solution], link_to_paper: bool):
        self.contest_id = contest_id
        self.user = user
        self.sol = sol
        self.link_to_paper = link_to_paper

    def __str__(self) -> str:
        if not self.sol:
            return '–'
        elif self.sol.points is None:
            return '?'
        return format_decimal(self.sol.points)

    def to_html(self) -> str:
        if not self.sol:
            return '<td>–'
        elif self.sol.points is None:
            points = '<span class="unknown">?</span>'
        else:
            points = format_decimal(self.sol.points)

        if not self.link_to_paper:
            return f'<td>{points}'
        elif self.sol.final_feedback_obj:
            url = mo.web.util.org_paper_link(self.contest_id, None, self.user, self.sol.final_feedback_obj)
            return f'<td><a href="{url}" title="Zobrazit finální opravu">{points}</a>'
        elif self.sol.final_submit_obj:
            url = mo.web.util.org_paper_link(self.contest_id, None, self.user, self.sol.final_submit_obj)
            return f'<td><a href="{url}" title="Zobrazit finální řešení od účastníka">{points}</a>'
        else:
            return f'<td>{points}'  # no paper no link


class ScoreEditForm(FlaskForm):
    submit = wtforms.SubmitField("Uložit zjednoznačnění")


class ScoreSnapshotForm(FlaskForm):
    note = wtforms.TextField("Poznámka k verzi (pro organizátory)")
    submit_snapshot = wtforms.SubmitField("Uložit současnou verzi")


@app.route('/org/contest/r/<int:round_id>/score')
@app.route('/org/contest/r/<int:round_id>/score/edit', methods=('GET', 'POST'), endpoint="org_score_edit")
@app.route('/org/contest/r/<int:round_id>/h/<int:hier_id>/score')
@app.route('/org/contest/r/<int:round_id>/h/<int:hier_id>/score/edit', methods=('GET', 'POST'), endpoint="org_score_edit")
@app.route('/org/contest/c/<int:ct_id>/score')
@app.route('/org/contest/c/<int:ct_id>/score/edit', methods=('GET', 'POST'), endpoint="org_score_edit")
def org_score(round_id: Optional[int] = None, hier_id: Optional[int] = None, ct_id: Optional[int] = None):
    ctx = get_context(round_id=round_id, hier_id=hier_id, ct_id=ct_id)
    contest = ctx.contest
    round = ctx.round
    format = request.args.get('format', "")

    sess = db.get_session()

    if not ctx.rights.have_right(Right.view_contestants):
        raise werkzeug.exceptions.Forbidden()
    can_view_submits = ctx.rights.have_right(Right.view_submits)

    is_edit = request.endpoint == 'org_score_edit'
    if is_edit and not ctx.rights.have_right(Right.manage_contest):
        raise werkzeug.exceptions.Forbidden()

    score = Score(round.master, contest, ctx.hier_place)
    tasks = score.get_tasks()
    results = score.get_sorted_results()
    messages = score.get_messages()

    edit_form: Optional[ScoreEditForm] = None
    snapshot_form: Optional[ScoreSnapshotForm] = None

    if is_edit:
        edit_form = ScoreEditForm()
        if edit_form.validate_on_submit():
            count = 0
            for result in results:
                try:
                    score_suborder = int(request.form.get(f"suborder_{result.user.user_id}"))
                except ValueError:
                    score_suborder = None

                if score_suborder != result.pion.score_suborder:
                    app.logger.info(f"Změněno zjednoznačnění u soutěžícího #{result.user.user_id} v soutěži #{result.pion.contest_id}: {result.pion.score_suborder}->{score_suborder}")
                    mo.util.log(
                        type=db.LogType.participant,
                        what=result.user.user_id,
                        details={
                            'action': 'pion-changed-suborder',
                            'contest_id': result.pion.contest_id,
                            'old_score_suborder': result.pion.score_suborder,
                            'new_score_suborder': score_suborder
                        },
                    )
                    result.pion.score_suborder = score_suborder
                    count += 1

            if count > 0:
                sess.commit()
                flash('Změněno zjednoznačnění u ' + inflect_number(count, 'soutěžícího', 'soutěžících', 'soutěžících'), 'success')
            else:
                flash('Žádné změny k uložení', 'info')
            return redirect(ctx.url_for('org_score'))


    # Pro tvorbu odkazů na správné contesty ve výsledkovkách dělených kol
    all_subcontests: List[db.Contest] = sess.query(db.Contest).filter(
        db.Contest.round_id.in_(
            sess.query(db.Round.round_id).filter_by(master_round_id=round.master_round_id)
        )
    ).all()
    subcontest_id_map = {}
    for subcontest in all_subcontests:
        subcontest_id_map[(subcontest.round_id, subcontest.master_contest_id)] = subcontest.contest_id

    # Construct columns
    columns = [
        Column(key='order',         name='poradi',          title='Pořadí'),
        Column(key='status',        name='stav',            title='Stav (vítěz, …)', in_html=False),
        Column(key='participant',   name='ucastnik',        title='Účastník',       in_export=False),
        Column(key='first_name',    name='krestni',         title='Křestní jméno',  in_html=False),
        Column(key='last_name',     name='prijmeni',        title='Příjmení',       in_html=False),
        Column(key='email',         name='email',           title='E-mail',         in_html=False),
    ]
    if not ct_id:
        columns.append(Column(key='contest', name='oblast', title='Soutěžní ' + round.get_level().name))
    columns.extend([
        Column(key='pion_place',    name='soutezni_misto',  title='Soutěžní místo', in_html=False),
        Column(key='school',        name='skola',           title='Škola'),
        Column(key='grade',         name='rocnik',          title='Ročník'),
        Column(key='birth_year',    name='rok_narozeni',    title='Rok narození',   in_html=False),
    ])
    for task in tasks:
        title = task.code
        if contest:
            local_ct_id = subcontest_id_map[(task.round_id, contest.master_contest_id)]
            title = '<a href="{}">{}</a>'.format(
                url_for('org_contest_task', ct_id=local_ct_id, task_id=task.task_id),
                task.code
            )
            if ctx.rights.can_edit_points():
                title += ' <a href="{}" title="Editovat body" class="icon">✎</a>'.format(
                    url_for('org_contest_task_edit', ct_id=local_ct_id, task_id=task.task_id),
                )
        columns.append(Column(key=f'task_{task.task_id}', name=task.code, title=title))
    columns.append(Column(key='total_points', name='celkove_body', title='Celkové body'))
    if is_edit:
        columns.append(Column(key='suborder', name='zjednoznacneni_poradi', title='Zjednoznačnění'))
    columns.append(Column(key='order_key', name='order_key', title='Třídící klíč', in_html=False, in_export=False))

    # Construct rows
    table_rows = []
    for result in results:
        user, pant, pion = result.user, result.pant, result.pion
        school = pant.school_place
        local_pion_ct_id = subcontest_id_map[(round.round_id, pion.contest_id)]

        order_cell: Union[Cell, str]
        if result.successful or not score.want_successful:
            order_cell = OrderCell(result.order.place, result.order.span, result.order.continuation)
        else:
            order_cell = ""

        if result.winner:
            status = 'vítěz'
        elif result.successful:
            status = 'úspěšný'
        else:
            status = ""

        row = Row(keys={
            'order':        order_cell,
            'status':       status,
            'user':         user,
            'first_name':   user.first_name,
            'last_name':    user.last_name,
            'email':        user.email,
            'participant':  cell_pion_link(user, local_pion_ct_id, user.full_name()),
            'contest':      CellLink(pion.contest.place.name or "?", url_for('org_contest', ct_id=pion.contest_id)),
            'pion_place':   pion.place.name,
            'school':       CellLink(school.name or "?", url_for('org_place', id=school.place_id)),
            'grade':        pant.grade,
            'total_points': format_decimal(result.get_total_points()),
            'birth_year':   pant.birth_year,
            'order_key':    result._order_key,
            'suborder':     CellInput(f"suborder_{user.user_id}", pion.score_suborder, "number", attrs={"size": 6}),
        })

        sols = result.get_sols_map()
        for task in tasks:
            local_sol_ct_id = subcontest_id_map[(task.round_id, pion.contest_id)]
            row.keys[f'task_{task.task_id}'] = SolPointsCell(
                contest_id=local_sol_ct_id, user=user, sol=sols.get(task.task_id),
                link_to_paper=can_view_submits
            )
        if result.winner:
            row.html_attr = {"class": "winner", "title": "Vítěz"}
        elif result.successful:
            row.html_attr = {"class": "successful", "title": "Úspěšný řešitel"}
        table_rows.append(row)

    filename = f"vysledky_{round.year}-{round.category}-{round.level}"
    if contest:
        filename += f"_oblast_{contest.place.code or contest.place.place_id}"
    table = Table(
        id="vysledky",
        table_class="data full center",
        columns=columns,
        rows=table_rows,
        filename=filename,
        show_downlink=not is_edit,
    )

    group_rounds = round.get_group_rounds(True)
    group_rounds.sort(key=lambda r: r.round_code())

    snapshots_count = db.get_count(sess.query(db.ScoreTable).filter_by(contest_id=ct_id))

    if ctx.rights.have_right(Right.manage_contest) and ctx.contest:
        snapshot_form = ScoreSnapshotForm()

    if format == "":
        return render_template(
            'org_score.html',
            ctx=ctx,
            tasks=tasks,
            table=table, messages=messages,
            group_rounds=group_rounds,
            snapshots_count=snapshots_count,
            edit_form=edit_form, snapshot_form=snapshot_form,
        )
    else:
        return table.send_as(format, args=request.args)


class SetFinalScoretableForm(FlaskForm):
    scoretable_id = wtforms.IntegerField()
    submit_set_final = wtforms.SubmitField("Zveřejnit")
    submit_hide = wtforms.SubmitField("Skrýt")


@app.route('/org/contest/c/<int:ct_id>/score/snapshots', methods=('GET', 'POST'))
def org_score_snapshots(ct_id: int):
    ctx = get_context(ct_id=ct_id)
    assert ctx.contest

    if not ctx.rights.have_right(Right.view_contestants):
        raise werkzeug.exceptions.Forbidden()

    sess = db.get_session()
    scoretables = sess.query(db.ScoreTable).filter_by(contest_id=ct_id).all()

    snapshot_form: Optional[ScoreSnapshotForm] = None
    set_final_form: Optional[SetFinalScoretableForm] = None

    if ctx.rights.have_right(Right.manage_contest):
        snapshot_form = ScoreSnapshotForm()
        if snapshot_form.validate_on_submit() and snapshot_form.submit_snapshot.data:

            mo.jobs.score.schedule_snapshot_score(ctx.contest, g.user, snapshot_form.note.data)
            return redirect(url_for('org_jobs'))

        set_final_form = SetFinalScoretableForm()
        if set_final_form.validate_on_submit():
            found = False
            scoretable_id = set_final_form.scoretable_id.data
            for scoretable in scoretables:
                if scoretable.scoretable_id == scoretable_id:
                    found = True
                    break
            if found and set_final_form.submit_set_final:
                ctx.contest.scoretable_id = scoretable_id
                mo.jobs.score.schedule_export_score_to_mo_web(scoretable)
                mo.util.log(
                    type=db.LogType.contest,
                    what=ctx.contest.contest_id,
                    details={
                        'action': 'score-publish',
                        'scoretable_id': scoretable_id,
                    },
                )
                sess.commit()
                app.logger.info(f"Zveřejněna výsledková listina #{scoretable_id} pro soutěž #{ctx.contest.contest_id}")
                flash("Výsledková listina zveřejněna.", "success")
            elif set_final_form.submit_hide.data:
                ctx.contest.scoretable_id = None
                mo.util.log(
                    type=db.LogType.contest,
                    what=ctx.contest.contest_id,
                    details={
                        'action': 'score-hide',
                    },
                )
                sess.commit()
                app.logger.info(f"Skryta výsledková listina pro soutěž #{ctx.contest.contest_id}")
                flash("Výsledková listina skryta.", "success")
            else:
                flash("Neznámé ID výsledkové listiny.", "danger")
            return redirect(ctx.url_for('org_score_snapshots'))

    return render_template(
        'org_score_snapshots.html',
        ctx=ctx,
        scoretables=scoretables,
        set_final_form=set_final_form
    )


@app.route('/org/contest/c/<int:ct_id>/score/<int:scoretable_id>.pdf')
def org_score_snapshot_pdf(ct_id: int, scoretable_id: int):
    ctx = get_context(ct_id=ct_id)
    assert ctx.contest

    if not (ctx.rights.have_right(Right.view_contestants) or scoretable_id == ctx.contest.scoretable_id):
        raise werkzeug.exceptions.Forbidden()

    sess = db.get_session()

    scoretable = sess.query(db.ScoreTable).get(scoretable_id)
    if not scoretable or scoretable.contest_id != ct_id or not scoretable.pdf_file:
        raise werkzeug.exceptions.NotFound()

    return mo.web.util.send_score_pdf(scoretable)


@app.route('/org/contest/c/<int:ct_id>/score/<int:scoretable_id>')
def org_score_snapshot(ct_id: int, scoretable_id: int):
    ctx = get_context(ct_id=ct_id)
    assert ctx.contest

    if not (ctx.rights.have_right(Right.view_contestants) or scoretable_id == ctx.contest.scoretable_id):
        raise werkzeug.exceptions.Forbidden()

    format = request.args.get('format', "")
    sess = db.get_session()

    scoretable = sess.query(db.ScoreTable).get(scoretable_id)
    if not scoretable or scoretable.contest_id != ct_id:
        raise werkzeug.exceptions.NotFound()

    columns, table_rows = scoretable_construct(scoretable)
    # columns.append(Column(key='order_key', name='order_key', title='Třídící klíč'))

    filename = f"vysledky_{ctx.round.year}-{ctx.round.category}-{ctx.round.level}_oblast_{ctx.contest.place.code or ctx.contest.place.place_id}"
    table = Table(
        id="vysledky",
        table_class="data full center",
        columns=columns,
        rows=table_rows,
        filename=filename,
    )

    if format == "":
        return render_template(
            'org_score_snapshot.html',
            ctx=ctx,
            table=table,
            scoretable=scoretable,
        )
    else:
        return table.send_as(format)