Skip to content
Snippets Groups Projects
Commit 008b6140 authored by Martin Mareš's avatar Martin Mareš
Browse files

Import dozoru

Zobecnil jsem modul na importování, ale ještě se ne úplně pěkně opakují
kusy webového rozhraní. Až budeme přidávat další importy, bude to chtít
vyřešit.
parent 397978d1
No related branches found
No related tags found
No related merge requests found
from dataclasses import dataclass
import io
import re
from typing import List, Optional, Any, Dict
from typing import List, Optional, Any, Dict, Callable, Type, TypeVar
import mo.csv
import mo.db as db
import mo.rights
import mo.util
RowType = TypeVar('RowType', bound=mo.csv.Row)
@dataclass
class ContestImportRow(mo.csv.Row):
......@@ -21,12 +23,21 @@ class ContestImportRow(mo.csv.Row):
kod_oblasti: str = ""
@dataclass
class ProctorImportRow(mo.csv.Row):
email: str = ""
krestni: str = ""
prijmeni: str = ""
kod_mista: str = ""
class Import:
line_errors: List[str]
errors: List[str]
user: db.User
round: Optional[db.Round]
contest: Optional[db.Contest]
round: db.Round
place_cache: Dict[str, db.Place]
school_place_cache: Dict[str, db.Place]
rr: Optional[mo.rights.Rights]
......@@ -81,6 +92,7 @@ class Import:
self.rr = mo.rights.Rights(self.user)
round = self.round
assert round is not None
self.rr.get_for(place, round.year, round.category, round.seq)
result = self.rr.have_right(mo.rights.Right.manage_contest)
self.place_rights_cache[place.place_id] = result
......@@ -146,7 +158,7 @@ class Import:
user = sess.query(db.User).filter_by(email=email).one_or_none()
if user:
if user.first_name != krestni or user.last_name != prijmeni:
return self.error(f'Účastník již registrován s odlišným jménem {user.first_name} {user.last_name}')
return self.error(f'Osoba již registrována s odlišným jménem {user.first_name} {user.last_name}')
else:
user = db.User(email=email, first_name=krestni, last_name=prijmeni)
sess.add(user)
......@@ -204,7 +216,7 @@ class Import:
return pion
def import_contest_row(self, contest: Optional[db.Contest], r: ContestImportRow):
def import_contest_row(self, r: ContestImportRow):
num_prev_errs = len(self.errors)
email = self.parse_email(r.email)
krestni = self.parse_name(r.krestni)
......@@ -232,7 +244,8 @@ class Import:
if part is None:
return
if contest:
if self.contest:
contest = self.contest
if oblast is not None and oblast.place_id != contest.place.place_id:
return self.error('Účastník soutěží mimo oblast, do které se importuje')
else:
......@@ -244,19 +257,58 @@ class Import:
self.find_or_create_participation(user, contest, misto)
def import_contest(self, round: db.Round, contest: Optional[db.Contest], path: str) -> bool:
self.round = round
def import_proctor_row(self, r: ProctorImportRow):
num_prev_errs = len(self.errors)
email = self.parse_email(r.email)
krestni = self.parse_name(r.krestni)
prijmeni = self.parse_name(r.prijmeni)
misto = self.parse_opt_place(r.kod_mista)
if misto is None:
return self.error('Kód místa je povinné uvést')
if (len(self.errors) > num_prev_errs
or email is None
or krestni is None
or prijmeni is None):
return
user = self.find_or_create_user(email, krestni, prijmeni)
if user is None:
return
sess = db.get_session()
round = self.round
assert round is not None
if (sess.query(db.UserRole)
.filter_by(user=user, place=misto, role=db.RoleType.dozor,
category=round.category, year=round.year, seq=round.seq)
.with_for_update()
.first()):
pass
else:
ur = db.UserRole(user=user, place=misto, role=db.RoleType.dozor,
category=round.category, year=round.year, seq=round.seq)
sess.add(ur)
sess.flush()
mo.util.log(
type=db.LogType.user_role,
what=ur.user_role_id,
details={'action': 'import', 'new': db.row2dict(ur)},
)
def generic_import(self, path: str, row_class: Type[RowType], process_row: Callable[['Import', RowType], None]) -> bool:
try:
with open(path) as file:
rows: List[ContestImportRow] = mo.csv.read(file=file, dialect='excel', row_class=ContestImportRow)
rows: List[RowType] = mo.csv.read(file=file, dialect='excel', row_class=row_class)
except Exception as e:
return self.error(f'Chybná struktura tabulky {e}')
line_num = 2
for row in rows:
self.line_errors = []
self.import_contest_row(contest, row)
process_row(self, row)
for err in self.line_errors:
self.errors.append(f"Řádek {line_num}: {err}")
if len(self.errors) >= 100:
......@@ -266,8 +318,26 @@ class Import:
return len(self.errors) == 0
def import_contest(self, round: db.Round, contest: Optional[db.Contest], path: str) -> bool:
self.round = round
self.contest = contest
return self.generic_import(path, ContestImportRow, Import.import_contest_row)
def contest_template() -> str:
def import_proctors(self, round: db.Round, path: str) -> bool:
self.round = round
self.contest = None
return self.generic_import(path, ProctorImportRow, Import.import_proctor_row)
def generic_template(row_class: Type[RowType]) -> str:
out = io.StringIO()
mo.csv.write(file=out, dialect='excel', row_class=ContestImportRow, rows=[])
mo.csv.write(file=out, dialect='excel', row_class=row_class, rows=[])
return out.getvalue()
def contest_template() -> str:
return generic_template(ContestImportRow)
def proctor_template() -> str:
return generic_template(ProctorImportRow)
......@@ -281,3 +281,45 @@ def make_contestant_table(round: db.Round, contest: Optional[db.Contest]) -> Tab
rows=rows,
filename='ucastnici',
)
@app.route('/org/contest/c/<int:id>/proctor-import', methods=('GET', 'POST'))
def org_proctor_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 dozoru: 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_proctors(contest.round, tmp_path):
mo.util.log(
type=db.LogType.contest,
what=contest.contest_id,
details={'action': 'import-proctors'}
)
db.get_session().commit()
flash('Dozor importován', '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_proctor_import.html',
contest=contest,
form=form,
errs=errs,
)
@app.route('/org/contest/import/sablona-dozor.csv')
def org_proctor_import_template():
out = mo.imports.proctor_template()
resp = app.make_response(out)
resp.content_type = 'text/csv; charset=utf=8'
return resp
......@@ -13,6 +13,7 @@
{% if can_manage %}
<p><a href='{{ url_for('org_contest_import', id=contest.contest_id) }}'>Importovat účastníky</a>
<p><a href='{{ url_for('org_contest_list', id=contest.contest_id) }}'>Seznam účastníků</a>
<p><a href='{{ url_for('org_proctor_import', id=contest.contest_id) }}'>Importovat dozor</a>
{% endif %}
<h3>Vaše práva k této soutěži</h3>
......
{% extends "base.html" %}
{% import "bootstrap/wtf.html" as wtf %}
{% block body %}
<h2>Soutěž {{ contest.round.round_code() }}: {{ contest.place.name }}</h2>
<a href='{{ url_for('org_contest', id=contest.contest_id) }}'>Zpět na soutěž</a>
{% if errs %}
<h3>Chyby při importu</h3>
<pre>
{% for e in errs %}
{{ e }}
{% endfor %}
</pre>
{% endif %}
<h3>Import dozoru</h3>
<p>Dozor na jednotlivých soutěžních místech můžete importovat ve <a href='{{ url_for('org_contest_import_help') }}'>formátu CSV</a> (FIXME)
podle <a href='{{ url_for('org_proctor_import_template') }}'>šablony</a>.
{{ wtf.quick_form(form, form_type='horizontal') }}
{% endblock %}
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment