Merge branch 'mj/points' into 'devel'

Prázdné protokoly

See merge request !134

db/db.ddl

@@ -265,6 +265,7 @@ CREATE TABLE solutions (
 	points		numeric(5,1)	DEFAULT NULL,
 	note		text		NOT NULL DEFAULT '',			-- komentář pro řešitele
 	org_note	text		NOT NULL DEFAULT '',			-- komentář viditelný jen organizátorům
+	is_empty	bool		NOT NULL DEFAULT false,			-- prázdný papír (v tomto případě je vždy points=0)
 	PRIMARY KEY (task_id, user_id)
 );
 
@@ -277,7 +278,8 @@ CREATE TABLE points_history (
 	participant_id	int		NOT NULL REFERENCES users(user_id),
 	points		numeric(5,1)	DEFAULT NULL,
 	points_by	int		NOT NULL REFERENCES users(user_id),		-- kdo přidělil body
-	points_at	timestamp with time zone	NOT NULL			-- a kdy
+	points_at	timestamp with time zone	NOT NULL,			-- a kdy
+	is_empty	bool		NOT NULL DEFAULT false
 );
 
 CREATE INDEX points_history_index ON points_history (task_id, participant_id);

db/upgrade-20230827.sql

+SET ROLE 'mo_osmo';
+
+ALTER TABLE solutions
+	ADD COLUMN is_empty	bool		NOT NULL DEFAULT false;
+
+ALTER TABLE points_history
+	ADD COLUMN is_empty	bool		NOT NULL DEFAULT false;

mo/db.py

@@ -748,6 +748,7 @@ class PointsHistory(Base):
     points = Column(Numeric)
     points_by = Column(Integer, ForeignKey('users.user_id'), nullable=False)
     points_at = Column(DateTime(True), nullable=False)
+    is_empty = Column(Boolean, server_default=text("false"), nullable=False)
 
     participant = relationship('User', primaryjoin='PointsHistory.participant_id == User.user_id')
     user = relationship('User', primaryjoin='PointsHistory.points_by == User.user_id')
@@ -764,6 +765,7 @@ class Solution(Base):
     points = Column(Numeric)
     note = Column(Text, nullable=False, server_default=text("''::text"))
     org_note = Column(Text, nullable=False, server_default=text("''::text"))
+    is_empty = Column(Boolean, server_default=text("false"), nullable=False)
 
     final_submit_obj = relationship('Paper', primaryjoin='Solution.final_submit == Paper.paper_id')
     final_feedback_obj = relationship('Paper', primaryjoin='Solution.final_feedback == Paper.paper_id')

mo/imports.py

+# Importování
+
 from dataclasses import dataclass, make_dataclass, field
 from decimal import Decimal
 from enum import auto
@@ -12,10 +14,12 @@ import mo.csv
 from mo.csv import FileFormat, MissingHeaderError
 import mo.db as db
 import mo.email
+import mo.points
+from mo.points import GenPoints, SolAction, SolActionError
 import mo.rights
 import mo.users
 import mo.util
-from mo.util import logger
+from mo.util import logger, assert_not_none
 from mo.util_format import format_decimal
 
 
@@ -60,6 +64,7 @@ class Import:
     new_user_ids: List[int]
     line_number: int = 0
     row_name: Optional[str] = None
+    to_log: List[str]
 
     def __init__(self, user: db.User):
         self.errors = []
@@ -68,6 +73,7 @@ class Import:
         self.school_place_cache = {}
         self.new_user_ids = []
         self.gatekeeper = mo.rights.Gatekeeper(user)
+        self.to_log = []
         self.user = user
 
     def error(self, msg: str) -> Any:
@@ -211,19 +217,15 @@ class Import:
             self.new_user_ids.append(user.user_id)
         return user
 
-    def parse_points(self, points_str: str, task: db.Task) -> Union[Decimal, str, None]:
+    def parse_gen_points(self, points_str: str, task: db.Task) -> Optional[GenPoints]:
         if points_str == "":
             return self.error(f'Body za úlohu {task.code} musí být vyplněny')
 
-        points_str = points_str.upper()
-        if points_str in ['X', '?']:
-            return points_str
-
-        pts, error = mo.util.parse_points(points_str, task, self.round)
-        if error:
-            return self.error(error)
+        gp = GenPoints.parse(points_str, task, self.round)
+        if gp.error:
+            return self.error(gp.error)
 
-        return pts
+        return gp
 
     def find_or_create_participant(self, user: db.User, year: int, school_id: Optional[int], birth_year: Optional[int], grade: Optional[str]) -> Optional[db.Participant]:
         try:
@@ -327,7 +329,10 @@ class Import:
         logger.info('Import: %s ze souboru %s: %s', self.log_msg_prefix, path, " ".join(args))
 
     def log_end(self) -> None:
-        args = [f'rows=#{self.cnt_rows}']
+        for msg in self.to_log:
+            logger.info(f'Import: {msg}')
+
+        args = [f'rows={self.cnt_rows}']
         for key, val in [
             ('users', self.cnt_new_users),
             ('p-ants', self.cnt_new_participants),
@@ -653,7 +658,7 @@ class PointsImportRowBase(mo.csv.Row):
 
 
 _SolutionDict = Dict[Tuple[int, int], Optional[db.Solution]]
-_TaskPoints = List[Tuple[db.Task, Union[Decimal, str, None]]]
+_TaskPoints = List[Tuple[db.Task, Optional[GenPoints]]]
 
 
 class PointsImport(Import):
@@ -730,13 +735,13 @@ class PointsImport(Import):
         user_id = self.parse_user_id(r.user_id)
         krestni = self.parse_name(r.krestni)
         prijmeni = self.parse_name(r.prijmeni)
-        task_points = [(task, self.parse_points(getattr(r, col), task)) for task, col in self.task_columns]
+        task_gen_points = [(task, self.parse_gen_points(getattr(r, col), task)) for task, col in self.task_columns]
 
         if (len(self.errors) > num_prev_errs
                 or user_id is None
                 or krestni is None
                 or prijmeni is None
-                or any(pts is None for task, pts in task_points)):
+                or any(gp is None for task, gp in task_gen_points)):
             return
 
         assert self.round is not None
@@ -762,85 +767,41 @@ class PointsImport(Import):
         else:
             contest = sess.query(db.Contest).filter_by(round=self.round, master_contest_id=pion.contest.contest_id).one()
 
+        # Práva není potřeba kontrolovat, stará se o to SolAction
         rights = self.gatekeeper.rights_for_contest(contest)
-        if not rights.can_edit_points():
-            return self.error('Nemáte právo na úpravu bodů')
 
         user = pion.user
         if user.first_name != krestni or user.last_name != prijmeni:
             return self.error('Neodpovídá ID a jméno soutěžícího')
 
-        self._add_del_solutions(user, sols, task_points, rights)
-        self._set_points(user, sols, task_points)
-
-    def _add_del_solutions(self,
-                           user: db.User,
-                           sols: _SolutionDict,
-                           task_points: _TaskPoints,
-                           rights: mo.rights.ContestRights) -> None:
-        user_id = user.user_id
-        sess = db.get_session()
+        for task, gp in task_gen_points:
+            assert gp is not None
 
-        for task, pts in task_points:
-            task_id = task.task_id
-            sol = sols[user_id, task_id]
-            if (sol is None) != (pts == 'X'):
-                if sol is None:
-                    if not self.allow_add_del:
-                        return self.error('Tento soutěžící úlohu neodevzdal')
-                    if not rights.can_upload_solutions():
-                        return self.error('Nemáte právo na zakládání nových řešení')
-                    sol = db.Solution(user_id=user_id, task_id=task_id)
-                    sols[user_id, task_id] = sol
-                    sess.add(sol)
-                    logger.info(f'Import: Založeno řešení user=#{user_id} task=#{task_id}')
-                    mo.util.log(
-                        type=db.LogType.participant,
-                        what=user_id,
-                        details={'action': 'solution-created', 'task': task_id},
+            sact = SolAction(
+                task=task,
+                user=user,
+                sol=sols[user.user_id, task.task_id],
+                gp=gp,
+                reason='import',
+                rights=rights,
+                to_log=self.to_log,
+                allow_add_del=self.allow_add_del,
             )
+
+            try:
+                added, deleted = sact.add_or_del()
+                if added:
                     self.cnt_add_sols += 1
-                elif pts == 'X':
-                    if not self.allow_add_del:
-                        return self.error('Tento soutěžící úlohu odevzdal')
-                    if sol.final_submit is not None or sol.final_feedback is not None:
-                        return self.error('Nelze smazat řešení, ke kterému existují odevzdané soubory')
-                    if not rights.can_upload_solutions():
-                        return self.error('Nemáte právo na mazání řešení')
-                    logger.info(f'Import: Smazáno řešení user=#{user_id} task=#{task_id}')
-                    mo.util.log(
-                        type=db.LogType.participant,
-                        what=user_id,
-                        details={'action': 'solution-removed', 'task': task_id},
-                    )
-                    sols[user_id, task_id] = None
-                    sess.delete(sol)
+                if deleted:
                     self.cnt_del_sols += 1
 
-    def _set_points(self,
-                    user: db.User,
-                    sols: _SolutionDict,
-                    task_points: _TaskPoints) -> None:
-        user_id = user.user_id
-        sess = db.get_session()
+                if sact.set_points():
+                    self.cnt_set_points += 1
 
-        for task, pts in task_points:
-            task_id = task.task_id
-            sol = sols[user_id, task_id]
-            if sol is None:
-                continue
+                sact.log_changes()
 
-            points = pts if isinstance(pts, Decimal) else None
-            if sol.points != points:
-                sol.points = points
-                sess.add(db.PointsHistory(
-                    task=task,
-                    participant_id=user_id,
-                    user=self.user,
-                    points_at=mo.now,
-                    points=points,
-                ))
-                self.cnt_set_points += 1
+            except SolActionError as e:
+                return self.error(str(e))
 
     def get_template(self) -> str:
         # Není to přímo PointsImportRowBase, ale její dynamicky generovaný potomek
@@ -858,12 +819,7 @@ class PointsImport(Import):
             )
             for task, col in self.task_columns:
                 sol = sols[user.user_id, task.task_id]
-                if sol is None:
-                    pts = 'X'
-                elif sol.points is None:
-                    pts = '?'
-                else:
-                    pts = format_decimal(sol.points)
+                pts = mo.points.format_sol_editable_points(sol, none_is_qmark=True)
                 setattr(row, col, pts)
             rows.append(row)
 

mo/place_level.py

+# Jména úrovní hierarchie míst
+
 from dataclasses import dataclass
 
 

mo/points.py

+# Pomocné funkce pro zacházení s hodnocením řešení
+
+from dataclasses import dataclass, field
+from decimal import Decimal, InvalidOperation
+from enum import auto
+from typing import Optional, Tuple, List
+
+import mo.db as db
+import mo.rights
+import mo.util
+from mo.util import assert_not_none
+import mo.util_format
+
+
+@dataclass
+class GenPoints:
+    have_solution: bool = True
+    is_empty: bool = False
+    points: Optional[Decimal] = None
+    error: Optional[str] = None
+
+    @staticmethod
+    def parse(
+        inp: Optional[str],
+        for_task: Optional[db.Task] = None,
+        for_round: Optional[db.Round] = None,
+    ) -> 'GenPoints':
+        """Zobecnění mo.util.parse_points(). Naparsuje generalizované body používané při editaci."""
+
+        inp = inp.upper() if inp is not None else None
+        gp = GenPoints()
+
+        if inp is None or inp == 'X':
+            gp.have_solution = False
+            return gp
+
+        if inp == "" or inp == '?':
+            pass
+        elif inp == 'P':
+            gp.is_empty = True
+            gp.points = Decimal(0)
+        else:
+            try:
+                gp.points = Decimal(inp.replace(',', '.'))
+            except InvalidOperation:
+                gp.error = f"Hodnota '{inp}' není číslo ani X nebo P"
+                return gp
+            gp.error = mo.util.check_points(gp.points, for_task, for_round)
+        return gp
+
+
+class SolActionError(RuntimeError):
+    pass
+
+
+@dataclass
+class SolAction:
+    task: db.Task
+    user: db.User
+    sol: Optional[db.Solution]
+    gp: GenPoints
+    reason: str
+    rights: mo.rights.RoundRights
+    to_log: List[str]
+    allow_add_del: bool = True
+    did_add_or_del: bool = False
+
+    def add_or_del(self) -> Tuple[bool, bool]:
+        if (self.sol is not None) == self.gp.have_solution:
+            return False, False
+
+        sess = db.get_session()
+        self.did_add_or_del = True
+
+        if self.sol is None:
+            if not self.allow_add_del:
+                raise SolActionError('Tento soutěžící úlohu neodevzdal')
+            if not (self.rights.can_upload_solutions() or self.rights.can_upload_feedback()):
+                raise SolActionError('Nemáte právo na zakládání nových řešení, můžete jen upravovat body')
+            self.sol = db.Solution(user=self.user, task=self.task)
+            sess.add(self.sol)
+            self.to_log.append(f'Založeno řešení user=#{self.user.user_id} task=#{self.task.task_id}')
+            mo.util.log(
+                type=db.LogType.participant,
+                what=self.user.user_id,
+                details={'action': 'solution-created', 'task': self.task.task_id, 'reason': self.reason},
+            )
+            return True, False
+        else:
+            if not self.allow_add_del:
+                raise SolActionError('Tento soutěžící úlohu odevzdal')
+            if self.sol.final_submit is not None or self.sol.final_feedback is not None:
+                raise SolActionError('Nelze smazat řešení, ke kterému existují odevzdané soubory')
+            if not self.rights.can_upload_solutions():
+                raise SolActionError('Nemáte právo na mazání řešení')
+            self.to_log.append(f'Smazáno řešení user=#{self.user.user_id} task=#{self.task.task_id}')
+            mo.util.log(
+                type=db.LogType.participant,
+                what=self.user.user_id,
+                details={'action': 'solution-removed', 'task': self.task.task_id, 'reason': self.reason},
+            )
+            sess.delete(self.sol)
+            self.sol = None
+            return False, True
+
+    def set_points(self) -> bool:
+        sol = self.sol
+        gp = self.gp
+
+        if sol is None:
+            return False
+        if gp.is_empty == sol.is_empty and gp.points == sol.points:
+            return False
+
+        if not self.rights.can_edit_points():
+            raise SolActionError('Nemáte právo hodnotit řešení')
+
+        sol.points = gp.points
+        sol.is_empty = gp.is_empty
+        sess = db.get_session()
+        sess.add(db.PointsHistory(
+            task=self.task,
+            participant_id=self.user.user_id,
+            user=mo.util.current_log_user,
+            points_at=mo.now,
+            points=gp.points,
+            is_empty=gp.is_empty,
+        ))
+        return True
+
+    def log_changes(self):
+        sess = db.get_session()
+        if self.sol and not self.did_add_or_del and sess.is_modified(self.sol):
+            changes = db.get_object_changes(self.sol)
+            mo.util.log(
+                type=db.LogType.participant,
+                what=self.user.user_id,
+                details={
+                    'action': 'solution-edit',
+                    'task': self.task.task_id,
+                    'changes': changes,
+                    'reason': self.reason,
+                },
+            )
+            self.to_log.append(f"Řešení user=#{self.user.user_id} task=#{self.task.task_id} modifikováno, změny: {changes}")
+
+
+def format_sol_editable_points(s: Optional[db.Solution], none_is_qmark: bool = False) -> str:
+    if s is None:
+        return 'X'
+    elif s.is_empty:
+        return 'P'
+    elif s.points is None:
+        return '?' if none_is_qmark else ""
+    else:
+        return assert_not_none(mo.util_format.format_decimal(s.points))

mo/score.py

+# Back-end pro generování výsledkových listin
+
 import decimal
 from fractions import Fraction
 from sqlalchemy import and_
@@ -230,6 +232,7 @@ class Score:
         sols: List[db.Solution] = sess.query(db.Solution).filter(
             db.Solution.user_id.in_(user_id_subq),
             db.Solution.task_id.in_(task_ids),
+            db.Solution.is_empty == False
         ).all()
         for sol in sols:
             if sol.user_id in self._results:

mo/submit.py

+# Back-end pro zpracování odevzdaných/opravených protokolů
+
 import datetime
 import multiprocessing
 import os

mo/util.py

@@ -154,26 +154,6 @@ def parse_points(
     return points, check_points(points, for_task, for_round)
 
 
-def parse_gen_points(
-    gen_points: Optional[str], for_task: Optional[db.Task] = None, for_round: Optional[db.Round] = None,
-) -> Tuple[str, Optional[decimal.Decimal], Optional[str]]:
-    """Zobecnění parse_points(). Naparsuje generalizované body používané při editaci.
-    Vrátí typ hodnocení, body (decimal.Decimal nebo None) a případný error."""
-    if gen_points is None or gen_points == 'X':
-        # Řešení nemá existovat
-        return 'X', None, None
-    elif gen_points == "" or gen_points == '?':
-        # Řešení má existovat, ale nemá přidělené body
-        return 'N', None, None
-    else:
-        # Řešení je ohodnoceno
-        try:
-            points = decimal.Decimal(gen_points.replace(',', '.'))
-        except decimal.InvalidOperation:
-            return "", decimal.Decimal(0), f"Hodnota '{gen_points}' není číslo ani X"
-        return 'P', points, check_points(points, for_task, for_round)
-
-
 def check_points(points: decimal.Decimal, for_task: Optional[db.Task] = None, for_round: Optional[db.Round] = None) -> Optional[str]:
     """Zkontroluje body. Pokud je vše ok, tak vrátí None, jinak vrátí text chyby."""
     if points < 0:

mo/util_format.py

-# Utils that do not depend on any other in mo (to avoid circular dependency)
+# Funkce pro formátování dat
+# Abychom se vyhnuli cyklickým závislostem, nezávisíme na žádných dalších modulech z mo.
 
 from datetime import datetime, timedelta
 import decimal

mo/web/__init__.py

+# Vstupní bod webového rozhraní
+
 from flask import Flask, request, g, session, redirect, url_for
 import flask.logging
 import flask.wrappers

mo/web/acct.py

+# Web: Základní funkce účtů
+
 import datetime
 import dateutil.tz
 from enum import Enum, auto

mo/web/api.py

+# Web: Interní API využívané JS funkcemi
+
 from flask import request
 from flask.json import jsonify
 from sqlalchemy import func

mo/web/doc.py

+# Web: Návody
+
 from flask import render_template
 
 from mo.web import app

mo/web/fields.py

+# Web: Formulářová políčka
+
 import decimal
 from typing import Optional
 import wtforms

mo/web/imports.py

+# Web: Importy
+
 from flask import render_template, g, redirect, url_for, flash, request
 from flask_wtf import FlaskForm
 import flask_wtf.file

mo/web/jinja.py

@@ -11,7 +11,9 @@ import urllib.parse
 import mo.config as config
 import mo.db as db
 import mo.place_level
+import mo.points
 from mo.rights import Right
+from mo.util import assert_not_none
 import mo.util_format as util_format
 from mo.web import app
 from mo.web.org_place import place_breadcrumbs
@@ -139,19 +141,19 @@ def user_flags(u: db.User) -> Markup:
 
 @app.template_filter()
 def sol_editable_points(s: Optional[db.Solution]) -> str:
-    if s is None:
-        return 'X'
-    elif s.points is None:
-        return ""
-    else:
-        return util_format.format_decimal(s.points)
+    return mo.points.format_sol_editable_points(s)
 
 
 @app.template_filter()
-def sol_display_points(s: Optional[db.Solution]) -> Union[str, Markup]:
+def sol_display_points(s: Optional[db.Solution], user: bool = False) -> Union[str, Markup]:
     if s is None:
         return '–'
+    elif s.is_empty:
+        return Markup('<span title="Prázdný protokol">∅</span>')
     elif s.points is None:
+        if user:
+            return '–'
+        else:
             return Markup('<span class="unknown">?</span>')
     else:
-        return util_format.format_decimal(s.points)
+        return assert_not_none(util_format.format_decimal(s.points))

mo/web/menu.py

+# Web: Menu
+
 from flask import request, url_for, g
 from typing import List, Optional
 

mo/web/misc.py

+# Web: Stránky, které nepatří jinam
+
 from flask import render_template, redirect, url_for, g
 
 from mo.web import app

mo/web/org.py

+# Web: Správa organizátorů
+
 from collections import defaultdict
 from dataclasses import dataclass, field
 from flask import render_template, redirect, url_for, request, flash, g