129 lines
4.2 KiB
Python
129 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
|
|
|
|
import datetime
|
|
import functools
|
|
import itertools
|
|
from mailbox import mbox, mboxMessage
|
|
import math
|
|
import re
|
|
from typing import Dict, List, Tuple
|
|
|
|
|
|
from submission import Submission, print_submissions
|
|
|
|
|
|
class Parser:
|
|
INFO_REGEX = re.compile(r"(\d{6}) \| (x\S*)\s*")
|
|
SUBMISSION_REGEX = re.compile(r"adresář:\s+\S*\/(\S*)\s*")
|
|
POINTS_REGEX = re.compile(r"\*\scelkový počet bodů\s+((\d|\.)*)\s*")
|
|
DATE_FORMAT = "%Y_%m%d_%H%M%S"
|
|
OFFSET_FOR_CORRECTION = datetime.timedelta(days=8)
|
|
|
|
@staticmethod
|
|
def get_match_from_mail(regex: re.Pattern, mail: mboxMessage) -> re.Match:
|
|
body = mail.get_payload()
|
|
match = regex.search(body)
|
|
if not match:
|
|
raise ValueError("invalid mail has been given")
|
|
|
|
return match
|
|
|
|
@staticmethod
|
|
def parse_info(mail: mboxMessage) -> Tuple[str, str]:
|
|
match = Parser.get_match_from_mail(Parser.INFO_REGEX, mail)
|
|
return match.group(1), match.group(2)
|
|
|
|
@staticmethod
|
|
def parse_submission(mail: mboxMessage) -> str:
|
|
match = Parser.get_match_from_mail(Parser.SUBMISSION_REGEX, mail)
|
|
return match.group(1)
|
|
|
|
@staticmethod
|
|
def parse_points(mail: mboxMessage) -> float:
|
|
match = Parser.get_match_from_mail(Parser.POINTS_REGEX, mail)
|
|
return float(match.group(1))
|
|
|
|
def __init__(self, path: str, deadline: str, correction: bool = False) -> None:
|
|
self.box = mbox(path)
|
|
|
|
self.deadline = datetime.datetime.strptime(deadline, Parser.DATE_FORMAT)
|
|
self.correction = correction
|
|
if correction:
|
|
# in case of correction pass the date of
|
|
# submitting review to IS for automatic computation
|
|
self.deadline = self.deadline.replace(hour=0, minute=0, second=0)
|
|
self.deadline += Parser.OFFSET_FOR_CORRECTION
|
|
|
|
def __get_submissions(self, hw_tag: str) -> Dict[str, List[Submission]]:
|
|
submissions = []
|
|
|
|
for mail in self.box.values():
|
|
uco, login = Parser.parse_info(mail)
|
|
path = Parser.parse_submission(mail)
|
|
points = Parser.parse_points(mail)
|
|
|
|
submissions.append(
|
|
Submission(uco, login, path, points, hw_tag, self.correction)
|
|
)
|
|
submissions[-1].set_late_tag(self.deadline)
|
|
|
|
submissions.sort(key=lambda e: (e.login, e.submitted_at))
|
|
return dict(
|
|
map(
|
|
lambda val: (val[0], list(val[1])),
|
|
itertools.groupby(submissions, key=lambda s: s.login),
|
|
)
|
|
)
|
|
|
|
def __filter(self, all_submissions: Dict[str, List[Submission]]) -> None:
|
|
for _, submissions in all_submissions.items():
|
|
length = len(submissions)
|
|
i = -1
|
|
while not submissions[i].submitted_before_deadline and i >= -length:
|
|
i -= 1
|
|
|
|
if i >= -length:
|
|
submissions[i].flag = "REVIEW"
|
|
|
|
def __correct_errors(self, all_submissions: Dict[str, List[Submission]]) -> None:
|
|
response = input("Do you wish to fix errors? ").strip()
|
|
if not response or response == "n":
|
|
return True
|
|
|
|
print("Choose indices of submissions that are to be reviewed")
|
|
for login, submissions in all_submissions.items():
|
|
response = input(f"==> Choose submission to review for {login}: ").strip()
|
|
if not response or response == "n":
|
|
continue
|
|
|
|
j = int(response)
|
|
for i, submission in enumerate(submissions):
|
|
if submission.flag == "REVIEW":
|
|
submission.flag = None
|
|
if i == j:
|
|
submission.flag = "REVIEW"
|
|
|
|
return False
|
|
|
|
def parse(self, hw_tag: str) -> List[Submission]:
|
|
def __reducer(lst, submission):
|
|
if submission.flag == "REVIEW":
|
|
lst.append(submission)
|
|
return lst
|
|
|
|
submissions = self.__get_submissions(hw_tag)
|
|
|
|
self.__filter(submissions)
|
|
print_submissions(submissions)
|
|
|
|
while not self.__correct_errors(submissions):
|
|
print_submissions(submissions)
|
|
|
|
result = []
|
|
for something in submissions.values():
|
|
functools.reduce(
|
|
__reducer, something, result,
|
|
)
|
|
return result
|
|
|