certs.py

# Implementace jobů na práci s diplomy

from dataclasses import dataclass
import os
from sqlalchemy import and_
from sqlalchemy.orm import joinedload
from typing import Dict, List, Optional, Any
import urllib.parse

import mo
import mo.config as config
import mo.db as db
from mo.jobs import TheJob, job_handler
import mo.util
from mo.util import logger
import mo.util_format
from mo.util_tex import tex_arg, run_tex, format_hacks, QREncoder


def schedule_create_certs(contest: db.Contest, for_user: db.User) -> int:
    place = contest.place

    the_job = TheJob()
    job = the_job.create(db.JobType.create_certs, for_user)
    job.description = f'Diplomy {contest.round.round_code()} {place.name_locative()}'
    job.in_json = {
        'contest_id': contest.contest_id,
    }
    the_job.submit()
    assert the_job.job_id is not None
    return the_job.job_id


@dataclass
class Cert:
    user: db.User
    school: db.Place
    type: db.CertType
    achievement: str
    sort_key: Any
    page_number: int = -1


class CertMaker:
    the_job: TheJob
    job: db.Job
    contest_id: int
    contest: db.Contest
    round: db.Round
    place: db.Place
    cset: db.CertSet
    scoretable: Optional[db.ScoreTable]

    certs: List[Cert]
    out_files: Dict[db.CertType, str]
    qr_encoder: QREncoder

    def __init__(self, the_job: TheJob):
        self.the_job = the_job
        self.job = the_job.job
        assert self.job.in_json is not None
        self.contest_id = self.job.in_json['contest_id']  # type: ignore

        sess = db.get_session()
        self.contest = (sess.query(db.Contest)
                        .options(joinedload(db.Contest.round))
                        .options(joinedload(db.Contest.place))
                        .options(joinedload(db.Contest.scoretable))
                        .get(self.contest_id)
                       )
        assert self.contest is not None
        self.round = self.contest.round
        self.place = self.contest.place
        self.scoretable = self.contest.scoretable

        self.cset = sess.query(db.CertSet).get(self.contest_id)
        assert self.cset is not None

        self.certs = []
        self.out_files = {}
        self.qr_encoder = QREncoder(f'{self.job.dir_path()}/qr')

    def plan(self) -> None:
        sess = db.get_session()
        pions_pants = (sess.query(db.Participation, db.Participant)
                       .select_from(db.Participation)
                       .join(db.Participation.contest).join(db.Contest.round)
                       .join(db.Participant, and_(db.Participant.user_id == db.Participation.user_id, db.Participant.year == db.Round.year))
                       .filter(db.Participation.contest_id == self.contest_id)
                       .filter(db.Participation.state == db.PartState.active)
                       .options(joinedload(db.Participation.user))
                       .options(joinedload(db.Participant.school_place))
                      )

        score_rows_by_user_id = {}
        if self.scoretable is not None:
            for row in self.scoretable.rows:
                uid: int = row['user_id']
                score_rows_by_user_id[uid] = row

        for pion, pant in pions_pants:
            user = pion.user
            row = score_rows_by_user_id.get(user.user_id)

            def add_cert(type: db.CertType, achievement: str, sort_key: Any) -> None:
                self.certs.append(Cert(
                    user=user,
                    school=pant.school_place,
                    type=type,
                    achievement=achievement,
                    sort_key=sort_key,
                ))

            # Účastnický list
            add_cert(db.CertType.participation, 'za účast', user.sort_key())

            # Diplom úspěšného řešitele
            if row is not None:
                order = row.get('order')
                if row['successful'] and order is not None:
                    if order['span'] == 1:
                        place = f"{order['place']}."
                    else:
                        place = f"{order['place']}.--{order['place'] + order['span'] - 1}."
                    add_cert(db.CertType.successful, f'za {place} místo', (order['place'], user.sort_key()))

            # Pochvalné uznání
            if row is not None and row.get('honorary_mention', False):
                add_cert(db.CertType.honorary_mention, 'za úplné vyřešení úlohy', user.sort_key())

    def make_certs(self, cert_type: db.CertType) -> None:
        certs = [cert for cert in self.certs if cert.type == cert_type]
        if not certs:
            return

        temp_dir = self.job.dir_path()
        name = cert_type.file_name()
        logger.debug(f'{self.the_job.log_prefix} Vytvářím certifikáty typu {name} v {temp_dir} ({len(certs)} listů)')

        certs.sort(key=lambda cert: cert.sort_key)
        self.make_tex_source(f'{temp_dir}/{name}.tex', certs)
        run_tex(temp_dir, f'{name}.tex')
        self.out_files[cert_type] = f'{temp_dir}/{name}.pdf'

    def make_tex_source(self, filename: str, certs: List[Cert]) -> None:
        with open(filename, 'w') as f:
            f.write('\\input certifikaty.tex\n\n')

            def attrs(adict: Dict[str, str]) -> None:
                for key, val in sorted(adict.items()):
                    f.write(f'\\def\\{key}' + tex_arg(val) + '\n')

            ga = {
                'kolo': db.round_type_names_local[self.round.round_type],
                'kat': self.round.category,
                'signerAname': self.cset.signer1_name,
                'signerAtitle': self.cset.signer1_title,
                'signerBname': self.cset.signer2_name,
                'signerBtitle': self.cset.signer2_title,
                'issueplace': self.cset.issue_place,
                'issuedate': self.cset.issue_date,
            }
            if self.round.round_type in (db.RoundType.okresni, db.RoundType.krajske):
                ga['oblast'] = self.place.name
            attrs(ga)
            f.write(format_hacks(self.cset.tex_hacks))

            for i, cert in enumerate(certs):
                qr_url = self._make_qr_url(cert)
                qr_file = self.qr_encoder.generate(qr_url)
                f.write('\n{\n')
                attrs({
                      'jmeno': cert.user.full_name(),
                      'skola': cert.school.name,
                      'uspech': cert.achievement,
                      'qrurl': qr_url,
                      'qrimg': os.path.basename(qr_file),
                })
                f.write('\\Cert' + cert.type.name.replace('_', '').title() + '\n')
                f.write('}\n')
                cert.page_number = i + 1

            f.write('\n\n\\bye\n')

    def _make_qr_url(self, cert: Cert) -> str:
        timestamp = int(self.job.started_at.timestamp())
        parts = [
            str(self.round.year),
            self.round.category,
            self.round.round_type.letter(),
            self.place.nuts or f'o{self.place.place_id}',
            cert.type.short_code(),
            str(cert.user.user_id),
            f'{timestamp:x}',
        ]
        return config.WEB_ROOT + 'cc/' + '/'.join(map(urllib.parse.quote, parts))

    def store_results(self) -> None:
        sess = db.get_session()
        conn = sess.connection()

        certs_dir = mo.util.data_dir('certs')
        out_dir = os.path.join(self.round.round_code_short(), str(self.contest.contest_id))
        full_dir = os.path.join(certs_dir, out_dir)
        os.makedirs(full_dir, exist_ok=True)

        # Nejdříve smažeme už zbytečné soubory s QR kódy
        self.qr_encoder.remove_all()

        # Najdeme všechny staré soubory s certifikáty
        old_files = [cfile.pdf_file for cfile in sess.query(db.CertFile).filter_by(cert_set_id=self.contest_id).all()]

        # Smažeme z DB staré certifikáty
        conn.execute(db.CertFile.__table__.delete().where(db.CertFile.cert_set_id == self.contest_id))
        conn.execute(db.Certificate.__table__.delete().where(db.Certificate.cert_set_id == self.contest_id))

        # Založíme nové soubory
        for ctype, out_file in self.out_files.items():
            file = mo.util.link_to_dir(out_file, full_dir, prefix=f'{ctype.file_name()}-', suffix='.pdf')
            sess.add(db.CertFile(
                cert_set_id=self.contest_id,
                type=ctype,
                pdf_file=os.path.join(out_dir, os.path.basename(file)),
            ))

        # Založíme nové certifikáty
        for cert in self.certs:
            sess.add(db.Certificate(
                cert_set_id=self.contest_id,
                user_id=cert.user.user_id,
                type=cert.type,
                achievement=cert.achievement,
                page_number=cert.page_number,
            ))

        # Aktualizujeme CertSet.
        # Timestamp výroby výsledkovky nastavujeme na start jobu, protože chceme,
        # aby případné změny provedené během provádění jobu byly považovány za novější.
        cset = self.cset
        cset.certs_issued_at = self.job.started_at
        cset.certs_issued_by = self.job.user_id
        cset.scoretable = self.scoretable

        # Zalogujeme změny CertSetu
        changes = db.get_object_changes(cset)
        mo.util.log(
            type=db.LogType.cert_set,
            what=self.contest.contest_id,
            details={'action': 'issued', 'changes': changes},
        )

        db.get_session().commit()

        # Nakonec smažeme staré soubory
        for old_file in old_files:
            mo.util.unlink_if_exists(os.path.join(certs_dir, old_file))


@job_handler(db.JobType.create_certs)
def handle_create_protocols(the_job: TheJob):
    cm = CertMaker(the_job)
    cm.plan()
    for ctype in db.CertType:
        cm.make_certs(ctype)
    cm.store_results()
    cm.job.result = f'Diplomy vytvořeny ({len(cm.certs)} ks).'