Skript freeze-score na dávkové mražení výsledkovek

bin/freeze-score

+#!/usr/bin/env python3
+
+import argparse
+import json
+import locale
+from pathlib import Path
+from shutil import copyfile
+from sqlalchemy.orm import joinedload
+from typing import List
+import PyPDF2
+
+import mo.db as db
+import mo.jobs.score
+import mo.util
+from mo.util import die, init_standalone
+
+
+def freeze(contests: List[db.Contest]) -> None:
+    for ct in contests:
+        mo.jobs.score.schedule_snapshot_score(ct, for_user=mo.db.get_system_user(), snapshot_note='freeze-score')
+
+
+def download(contests: List[db.Contest], out_name: str) -> None:
+    out_dir = Path(out_name)
+    out_dir.mkdir(exist_ok=True)
+    pdfs = []
+    jsons = []
+
+    for ct in contests:
+        out = out_dir / ct.place.get_code()
+        print(f'{out} ({ct.place.name}): ', end="")
+        st = sess.query(db.ScoreTable).filter_by(contest_id=ct.contest_id).order_by(db.ScoreTable.created_at.desc()).limit(1).one_or_none()
+        if st is None:
+            print('NONE')
+        else:
+            if ct.scoretable_id == st.scoretable_id:
+                print('official')
+            else:
+                print('unofficial')
+
+            pdf = out.with_suffix('.pdf')
+            copyfile(Path(mo.util.data_dir('score')) / st.pdf_file, pdf)
+            pdfs.append(pdf)
+
+            js = mo.jobs.score.get_web_json(st, ct)
+            jsons.append(js)
+            with open(out.with_suffix('.json'), 'w') as f:
+                json.dump(js, f, ensure_ascii=False, indent=4, sort_keys=True)
+
+    all_json = out_dir / 'all.json'
+    print(f'Merging JSONs to {all_json}')
+    with open(all_json, 'w') as f:
+        json.dump(jsons, f, ensure_ascii=False, indent=4, sort_keys=True)
+
+    all_pdf = out_dir / 'all.pdf'
+    print(f'Merging PDFs to {all_pdf}')
+    writer = PyPDF2.PdfWriter()
+    for pdf in pdfs:
+        reader = PyPDF2.PdfReader(pdf)
+        for pg in reader.pages:
+            writer.add_page(pg)
+    with open(all_pdf, 'wb') as f:
+        writer.write(f)
+
+
+def make_official(contests: List[db.Contest]) -> None:
+    for ct in contests:
+        print(f'{ct.place.code} ({ct.place.name}): ', end="")
+        st = sess.query(db.ScoreTable).filter_by(contest_id=ct.contest_id).order_by(db.ScoreTable.created_at.desc()).limit(1).one_or_none()
+        if st is None:
+            print('NONE')
+        else:
+            print('OK')
+            ct.scoretable_id = st.scoretable_id
+    sess.commit()
+
+
+parser = argparse.ArgumentParser(description='Dávkové zpracování zmražených výsledkových listin')
+
+cmds = parser.add_mutually_exclusive_group(required=True)
+cmds.add_argument('--freeze', default=False, action='store_true', help='zmrazí všechny výsledkové listiny')
+cmds.add_argument('--download', metavar='DIR', help='stáhne poslední zmraženou verzi (PDF i JSON)')
+cmds.add_argument('--make-official', default=False, action='store_true', help='prohlásí poslední zmraženou verzi za oficiální')
+
+parser.add_argument('--round', type=str, required=True, metavar='YY-C-S[p]', help='kód kola')
+
+args = parser.parse_args()
+init_standalone()
+sess = db.get_session()
+
+round_code = mo.util.RoundCode.parse(args.round)
+if round_code is None:
+    die("Chybná syntaxe kódu kola")
+round = mo.util.get_round_by_code(round_code)
+if round is None:
+    die("Kolo s tímto kódem neexistuje!")
+
+contests = sess.query(db.Contest).filter_by(round=round).options(joinedload(db.Contest.place)).all()
+contests.sort(key=lambda ct: locale.strxfrm(ct.place.name))
+
+if args.freeze:
+    freeze(contests)
+if args.download:
+    download(contests, args.download)
+if args.make_official:
+    make_official(contests)