org_contest.py

from flask import render_template, g, redirect, url_for, flash, request
from flask_wtf import FlaskForm
import flask_wtf.file
import locale
import os
import secrets
from sqlalchemy.orm import joinedload
from typing import List, Tuple, Optional, Sequence
import werkzeug.exceptions
import wtforms

import mo
import mo.csv
import mo.db as db
import mo.imports
import mo.rights
import mo.util
from mo.web import app
from mo.web.table import Table, Column, cell_place_link, cell_user_link, cell_email_link
import wtforms.validators as validators


class ImportForm(FlaskForm):
    file = flask_wtf.file.FileField("Soubor", validators=[flask_wtf.file.FileRequired()])
    submit = wtforms.SubmitField('Importovat')


@app.route('/org/contest/')
def org_contest_root():
    sess = db.get_session()

    rounds = sess.query(db.Round).filter_by(year=mo.current_year).order_by(db.Round.year, db.Round.category, db.Round.seq)
    return render_template('org_contest_root.html', rounds=rounds, level_names=mo.db.place_level_names)


def get_round(id: int) -> db.Round:
    round = db.get_session().query(db.Round).get(id)
    if not round:
        raise werkzeug.exceptions.NotFound()
    return round


def get_round_rr(id: int, right_needed: Optional[mo.rights.Right]) -> Tuple[db.Round, mo.rights.Rights]:
    round = get_round(id)

    rr = mo.rights.Rights(g.user)
    rr.get_for_round(round)

    if not (right_needed is None or rr.have_right(right_needed)):
        raise werkzeug.exceptions.Forbidden()

    return round, rr


@app.route('/org/contest/r/<int:id>/')
def org_round(id: int):
    sess = db.get_session()
    round, rr = get_round_rr(id, None)

    contests = (sess.query(db.Contest)
                .filter_by(round=round)
                .options(joinedload(db.Contest.place))
                .all())

    contests.sort(key=lambda c: locale.strxfrm(c.place.name))

    return render_template(
        'org_round.html',
        round=round,
        contests=contests,
        level_names=mo.db.place_level_names,
        can_manage=rr.have_right(mo.rights.Right.manage_contest),
    )


@app.route('/org/contest/r/<int:id>/list')
def org_round_list(id: int):
    round, rr = get_round_rr(id, mo.rights.Right.manage_contest)
    format = request.args.get('format', "")

    table = make_contestant_table(round, None)

    if format == "":
        return render_template(
            'org_round_list.html',
            round=round,
            table=table,
        )
    else:
        return table.send_as(format)


@app.route('/org/contest/r/<int:id>/import', methods=('GET', 'POST'))
def org_round_import(id: int):
    round, rr = get_round_rr(id, mo.rights.Right.manage_contest)

    form = ImportForm()
    errs = []
    if form.validate_on_submit():
        tmp_name = secrets.token_hex(16) + '.csv'
        tmp_path = os.path.join(app.instance_path, 'imports', tmp_name)
        form.file.data.save(tmp_path)
        app.logger.info('Import: Zpracovávám soubor %s pro round=%s, uid=%s', tmp_name, round.round_code(), g.user.user_id)

        imp = mo.imports.Import(g.user)
        if imp.import_contest(round, None, tmp_path):
            mo.util.log(
                type=db.LogType.round,
                what=round.round_id,
                details={'action': 'import'}
            )
            db.get_session().commit()
            flash('Účastníci importováni', 'success')
            return redirect(url_for('org_round', id=round.round_id))
        else:
            flash('Došlo k chybě při importu (detaily níže)', 'danger')
            errs = imp.errors

    return render_template(
        'org_round_import.html',
        round=round,
        form=form,
        errs=errs,
    )


def get_contest(id: int) -> db.Contest:
    contest = (db.get_session().query(db.Contest)
               .options(joinedload(db.Contest.place),
                        joinedload(db.Contest.round))
               .get(id))
    if not contest:
        raise werkzeug.exceptions.NotFound()
    return contest


def get_contest_rr(id: int, right_needed: Optional[mo.rights.Right]) -> Tuple[db.Contest, mo.rights.Rights]:
    contest = get_contest(id)

    rr = mo.rights.Rights(g.user)
    rr.get_for_contest(contest)

    if not (right_needed is None or rr.have_right(right_needed)):
        raise werkzeug.exceptions.Forbidden()

    return contest, rr


@app.route('/org/contest/c/<int:id>')
def org_contest(id: int):
    contest, rr = get_contest_rr(id, None)

    return render_template(
        'org_contest.html',
        contest=contest,
        rights=sorted(rr.current_rights, key=lambda r: r. name),
        can_manage=rr.have_right(mo.rights.Right.manage_contest),
    )


@app.route('/org/contest/c/<int:id>/import', methods=('GET', 'POST'))
def org_contest_import(id: int):
    contest, rr = get_contest_rr(id, mo.rights.Right.manage_contest)

    form = ImportForm()
    errs = []
    if form.validate_on_submit():
        tmp_name = secrets.token_hex(16) + '.csv'
        tmp_path = os.path.join(app.instance_path, 'imports', tmp_name)
        form.file.data.save(tmp_path)
        app.logger.info('Import: Zpracovávám soubor %s pro contest_id=%s, uid=%s', tmp_name, contest.contest_id, g.user.user_id)

        imp = mo.imports.Import(g.user)
        if imp.import_contest(contest.round, contest, tmp_path):
            mo.util.log(
                type=db.LogType.contest,
                what=contest.contest_id,
                details={'action': 'import'}
            )
            db.get_session().commit()
            flash('Účastníci importováni', 'success')
            return redirect(url_for('org_contest', id=contest.contest_id))
        else:
            flash('Došlo k chybě při importu (detaily níže)', 'danger')
            errs = imp.errors

    return render_template(
        'org_contest_import.html',
        contest=contest,
        form=form,
        errs=errs,
    )


@app.route('/org/contest/import/help.html')
def org_contest_import_help():
    return render_template('org_contest_import_help.html')


@app.route('/org/contest/import/sablona.csv')
def org_contest_import_template():
    out = mo.imports.contest_template()
    resp = app.make_response(out)
    resp.content_type = 'text/csv; charset=utf=8'
    return resp


@app.route('/org/contest/c/<int:id>/ucastnici')
def org_contest_list(id: int):
    contest, rr = get_contest_rr(id, mo.rights.Right.manage_contest)
    format = request.args.get('format', "")

    table = make_contestant_table(contest.round, contest)

    if format == "":
        return render_template(
            'org_contest_list.html',
            contest=contest,
            table=table,
        )
    else:
        return table.send_as(format)


contest_list_columns = (
    Column(key='first_name',    name='krestni',     title='Křestní jméno'),
    Column(key='last_name',     name='prijmeni',    title='Příjmení'),
    Column(key='email',         name='email',       title='E-mail'),
    Column(key='school',        name='skola',       title='Škola'),
    Column(key='school_code',   name='kod_skoly',   title='Kód školy'),
    Column(key='grade',         name='rocnik',      title='Ročník'),
    Column(key='born_year',     name='rok_naroz',   title='Rok naroz.'),
    Column(key='place_code',    name='kod_soutez_mista',  title='Sout. místo'),
    Column(key='status',        name='stav',        title='Stav'),
)


def make_contestant_table(round: db.Round, contest: Optional[db.Contest]) -> Table:
    query = (db.get_session()
             .query(db.Participation, db.Participant, db.Contest)
             .select_from(db.Participation)
             .join(db.Participant, db.Participant.user_id == db.Participation.user_id)
             .filter(db.Participant.year == round.year))
    if contest:
        query = query.join(db.Contest, db.Contest.contest_id == contest.contest_id)
    else:
        query = query.filter(db.Contest.round == round)
        query = query.options(joinedload(db.Contest.place))
    query = query.filter(db.Participation.contest_id == db.Contest.contest_id)
    query = query.options(joinedload(db.Participation.user),
                          joinedload(db.Participation.place),
                          joinedload(db.Participant.school_place))

    ctants = query.all()

    rows: List[dict] = []
    for pion, pant, ct in ctants:
        u = pion.user
        rows.append({
            'sort_key': (locale.strxfrm(u.last_name), locale.strxfrm(u.first_name), u.user_id),
            'first_name': cell_user_link(u, u.first_name),
            'last_name': cell_user_link(u, u.last_name),
            'email': cell_email_link(u),
            'school': pant.school_place.name,
            'school_code': cell_place_link(pant.school_place, pant.school_place.get_code()),
            'grade': pant.grade,
            'born_year': pant.birth_year,
            'region_code': cell_place_link(ct.place, ct.place.get_code()),
            'place_code': cell_place_link(pion.place, pion.place.get_code()),
            'status': pion.state.friendly_name(),
        })

    rows.sort(key=lambda r: r['sort_key'])

    cols: Sequence[Column] = contest_list_columns
    if not contest:
        cols = list(cols) + [Column(key='region_code', name='kod_oblasti', title='Oblast')]

    return Table(
        columns=cols,
        rows=rows,
        filename='ucastnici',
    )