Select Git revision
org_score.py
-
Jan Prachař authoredJan Prachař authored
org_certs.py 12.61 KiB
# Web: Certifikáty
from datetime import datetime
from flask import render_template, g, redirect, url_for
from flask.helpers import send_file, flash
from flask_wtf import FlaskForm
import flask_wtf.file
from markupsafe import Markup
import os
import pikepdf
from pikepdf.models.metadata import encode_pdf_date
from sqlalchemy.orm import joinedload
from tempfile import NamedTemporaryFile
from typing import Tuple, Optional, Dict
import werkzeug.exceptions
import wtforms
from wtforms import validators
import mo
import mo.db as db
import mo.email
from mo.jobs.certs import schedule_create_certs, DesignParams, BackgroundType
from mo.rights import Right
import mo.submit
import mo.util
from mo.util import logger, merge_objects
from mo.web import app
from mo.web.org_contest import get_context
import mo.web.fields as mo_fields
# Využívá se i z účastnické části webu
def send_certificate(ct_id: int, cert_type: str, user_filename: str, user_id: Optional[int] = None):
try:
typ = db.CertType.coerce(cert_type)
except ValueError:
raise werkzeug.exceptions.NotFound()
if user_filename != typ.file_name() + '.pdf':
raise werkzeug.exceptions.NotFound()
sess = db.get_session()
cfile = sess.query(db.CertFile).get((ct_id, typ))
if cfile is None:
raise werkzeug.exceptions.NotFound()
file = os.path.join(mo.util.data_dir('certs'), cfile.pdf_file)
try:
stat = os.stat(file)
except FileNotFoundError:
logger.error(f'Certifikát {file} je v DB, ale soubor neexistuje')
raise werkzeug.exceptions.NotFound()
if user_id is None:
return send_file(file, mimetype='application/pdf')
cert = sess.query(db.Certificate).get((ct_id, user_id, typ))
if cert is None:
raise werkzeug.exceptions.NotFound()
try:
with pikepdf.open(file, attempt_recovery=False) as src:
dst = pikepdf.new()
dst.pages.append(src.pages[cert.page_number - 1])
dst.docinfo['/Title'] = f'Matematická Olympiáda – {typ.friendly_name()}'
dst.docinfo['/Creator'] = 'Odevzdávací Systém Matematické Olympiády'
dst.docinfo['/CreationDate'] = encode_pdf_date(datetime.fromtimestamp(stat.st_mtime).astimezone())
tmp_file = NamedTemporaryFile(dir=mo.util.data_dir('tmp'), prefix='cert-')
dst.save(tmp_file.name)
except pikepdf.PdfError as e:
logger.error(f'Chyba při zpracování certifikátu {file}: {e}')
raise werkzeug.exceptions.InternalServerError()
return send_file(open(tmp_file.name, 'rb'), mimetype='application/pdf')
def validate_background(form: 'CertSetForm', field: wtforms.Field) -> None:
if not field.data:
return
file = field.data.stream
submitter = mo.submit.Submitter()
try:
submitter.check_certificate_background(file.name)
except mo.submit.SubmitException as e:
raise wtforms.ValidationError(str(e))
class SpaceField(wtforms.FloatField):
def __init__(self, label: str, **kwargs):
super().__init__(label,
validators=[validators.InputRequired(),
validators.NumberRange(-10, 200, "Hodnota musí být mezi %(min)s a %(max)s")],
**kwargs)
default_design = DesignParams()
class CertSetForm(FlaskForm):
signer1_name = mo_fields.String("1. podepisující: jméno")
signer1_title = mo_fields.String("1. podepisující: funkce", render_kw={'placeholder': 'např. Předseda okresní komise MO'})
signer2_name = mo_fields.String("2. podepisující: jméno")
signer2_title = mo_fields.String("2. podepisující: funkce", render_kw={'placeholder': 'např. Ředitel školy'})
issue_place = mo_fields.String("Místo vydání")
issue_date = mo_fields.String("Datum vydání", render_kw={'placeholder': 'např. 3. března 2025'})
background_type = wtforms.SelectField(choices=BackgroundType.choices(), coerce=BackgroundType.coerce)
upload_background = flask_wtf.file.FileField(validators=[validate_background],
render_kw={'accept': 'application/pdf'},
description="Zde můžete nahrát obrázek ve formátu PDF, který se použije jako pozadí diplomu.")
space1 = SpaceField("Mezera 1", description=f"Mezera mezi horním okrajem a hlavičkou (default: {default_design.space1}).")
space2 = SpaceField("Mezera 2", description=f"Mezera mezi hlavičkou a jménem (default: {default_design.space2}).")
space3 = SpaceField("Mezera 3", description=f"Mezera mezi jménem a umístěním (default: {default_design.space3}).")
space4 = SpaceField("Mezera 4", description=f"Mezera mezi umístěním a podpisy (default: {default_design.space4}).")
space5 = SpaceField("Mezera 5", description=f"Mezera mezi podpisy a patičkou (default: {default_design.space5}).")
space6 = SpaceField("Mezera 6", description=f"Mezera mezi patičkou a dolním okrajem (default: {default_design.space6}).")
logo_visible = wtforms.BooleanField("Logo MO a JČMF v patičce")
tex_hacks = mo_fields.String("Nastavení sazby") # description se nastavuje později
generate = wtforms.SubmitField("Vytvořit diplomy")
save = wtforms.SubmitField("Pouze uložit nastavení")
delete = wtforms.SubmitField("Smazat diplomy")
def osmo_validate(self, cset) -> bool:
ok = True
if self.background_type.data == BackgroundType.custom and self.upload_background.data is None and not cset.background_file:
self.upload_background.errors.append('Nahrajte obrázek na pozadí.') # FIXME: typing
ok = False
if all(getattr(self, key).data >= 0 for key in DesignParams.SPACE_PARAMS):
self.space1.errors.append('Alespoň jedna z mezer musí být pružná.') # FIXME: typing
ok = False
return ok
@app.route('/org/contest/c/<int:ct_id>/certificates', methods=('GET', 'POST'))
def org_certificates(ct_id: int):
ctx = get_context(ct_id=ct_id, right_needed=Right.view_contestants)
assert ctx.contest and ctx.master_contest
group_rounds = ctx.round.get_group_rounds(True)
group_rounds.sort(key=lambda r: r.round_code())
contest = ctx.master_contest
ct_id = contest.contest_id
can_change = ctx.rights.have_right(Right.manage_contest) and ctx.master_round.round_type != db.RoundType.other
sess = db.get_session()
cset = (sess.query(db.CertSet)
.with_for_update()
.filter_by(contest=contest)
.one_or_none())
if cset is None:
new_cset = True
cset = db.CertSet(
contest_id=ct_id,
)
dparams = DesignParams(issue_date=mo.now.strftime('%d. %B %Y'))
else:
new_cset = False
dparams = DesignParams.from_json(cset.design_params)
if can_change:
form = CertSetForm(obj=merge_objects(cset, dparams))
form.tex_hacks.description = Markup("Speciální nastavení sazby diplomů (viz <a href='" + url_for('doc_admin') + "'>návod</a>)")
if not ctx.rights.have_right(Right.edit_tex_hacks):
del form.tex_hacks
if new_cset:
del form.delete
else:
form = None
if form:
if not form.is_submitted():
pass
elif not form.validate() or not form.osmo_validate(cset):
flash('V nastavení diplomů byly nalezeny chyby.', 'danger')
elif not new_cset and form.delete.data:
sess.delete(cset)
mo.util.log(
type=db.LogType.cert_set,
what=contest.contest_id,
details={'action': 'delete', 'cert_set': db.row2dict(cset), 'reason': 'web'},
)
sess.commit()
app.logger.info(f'Smazány diplomy pro soutěž #{contest.contest_id}')
flash('Diplomy smazány.', 'success')
return redirect(ctx.url_for('org_certificates'))
elif form.generate.data or form.save.data:
form.populate_obj(cset)
form.populate_obj(dparams)
cset.design_params = dparams.to_json()
if dparams.background_type == BackgroundType.custom:
if form.upload_background.data:
old_background = cset.background_file
out_dir = cset.dir_path()
cset.background_file = mo.util.link_to_dir(form.upload_background.data.stream.name,
out_dir,
base_dir=mo.util.data_dir('certs'),
prefix='background-',
suffix='.pdf',
make_dirs=True)
app.logger.info(f'Nahráno pozadí diplomů {cset.background_file}')
else:
old_background = None
else:
old_background = cset.background_file
cset.background_file = None
changes = None
if new_cset:
sess.add(cset)
mo.util.log(
type=db.LogType.cert_set,
what=contest.contest_id,
details={'action': 'new', 'cert_set': db.row2dict(cset), 'reason': 'web'},
)
app.logger.info(f'Založeny diplomy pro soutěž #{contest.contest_id}')
sess.commit()
elif sess.is_modified(cset):
cset.changed_at = mo.now
changes = db.get_object_changes(cset)
mo.util.log(
type=db.LogType.cert_set,
what=contest.contest_id,
details={'action': 'edit', 'changes': changes, 'reason': 'web'},
)
app.logger.info(f'Upraveno nastavení diplomů pro soutěž #{contest.contest_id}')
sess.commit()
if old_background:
mo.util.unlink_if_exists(os.path.join(mo.util.data_dir('certs'), old_background))
if form.generate.data:
if cset.job is not None and cset.job.is_active():
flash('Počkejte, až doběhne předchozí dávka na tvorbu diplomů.', 'danger')
else:
cset.job_id = schedule_create_certs(contest, g.user)
sess.commit()
return redirect(url_for('org_job_wait', id=cset.job_id, back=ctx.url_for('org_certificates')))
else:
if changes is not None:
flash('Nastavení uloženo.', 'success')
else:
flash('Žádné změny k uložení.', 'info')
return redirect(ctx.url_for('org_certificates'))
pions = sess.query(db.Participation).filter_by(contest_id=ct_id).options(joinedload(db.Participation.user)).all()
pions.sort(key=lambda p: p.user.sort_key())
users_pions = [(p.user, p) for p in pions]
cert_files_by_type: Dict[db.CertType, Optional[db.CertFile]] = {}
for cf in sess.query(db.CertFile).filter_by(cert_set_id=ct_id).all():
cert_files_by_type[cf.type] = cf
certs_by_uid_type: Dict[Tuple[int, db.CertType], db.Certificate] = {}
for c in sess.query(db.Certificate).filter_by(cert_set_id=ct_id).all():
certs_by_uid_type[c.user_id, c.type] = c
return render_template(
'org_certificates.html',
ctx=ctx,
group_rounds=group_rounds,
form=form,
cset=cset,
users_pions=users_pions,
cert_files_by_type=cert_files_by_type,
certs_by_uid_type=certs_by_uid_type,
settings_changed=(cset.changed_at is not None and (cset.certs_issued_at is None or cset.certs_issued_at < cset.changed_at)),
scoretable_changed=(cset.scoretable != contest.scoretable),
form_has_errors=form and form.is_submitted(),
)
@app.route('/org/contest/c/<int:ct_id>/certificate/<cert_type>/all/<filename>')
@app.route('/org/contest/c/<int:ct_id>/certificate/<cert_type>/<int:user_id>/<filename>')
def org_cert_file(ct_id: int, cert_type: str, filename: str, user_id: Optional[int] = None):
ctx = get_context(ct_id=ct_id, right_needed=Right.view_contestants)
assert ctx.contest and ctx.master_contest
return send_certificate(ct_id, cert_type, filename, user_id)
# URL je explicitně uvedeno v mo.jobs.certs.Cert._make_qr_url
@app.route('/cc/<int:year>/<cat>/<round_type>/<place>/<cert_type_short>/<int:user_id>/<time_code>')
def cert_check(year: int, cat: str, round_type: str, place: str, cert_type_short: str, user_id: int, time_code: str):
return "Not implemented yet."