pushee/kontr_emails.py

#!/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
import json


from submission import Submission, print_submissions


class Parser:
    INFO_REGEX = re.compile(r"<td>(\d{6}) \| <code>(x\S*)</code>\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_mail(mail: mboxMessage) -> str:
        body = None
        for payload in mail.get_payload():
            if payload["Content-Type"].startswith("text/html"):
                body = payload.get_payload()

        return body

    @staticmethod
    def get_match_from_mail(regex: re.Pattern, mail: str) -> re.Match:
        match = regex.search(mail)
        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), match.group(2)

    @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, details: Dict, reviewer: str, homework: str) -> None:
        self.homework = homework

        self.from_ids = "submissions" in details
        if self.from_ids:
            self.submissions = details["submissions"]
        else:
            self.box = mbox(details["mbox_path"])
        self.reviewer = reviewer

        self.deadline = datetime.datetime.strptime(
            details["deadline"], Parser.DATE_FORMAT
        )
        self.correction = details["correction"]
        if self.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]]:
        mails = []
        if self.from_ids:
            print(self.submissions)
            for submission_id in self.submissions:
                with open(
                    f"/home/kontr/kontr/_tmp_/pb071/{self.homework}/{submission_id}/teacher_email",
                    "r",
                ) as f:
                    mails.append(f.read())
        else:
            for mail in self.box.values():
                if self.reviewer not in mail["subject"]:
                    continue
                mails.append(Parser.get_mail(mail))

        submissions = []
        for text in mails:
            uco, login = Parser.parse_info(text)
            path, submission_id = Parser.parse_submission(text)
            points = Parser.parse_points(text)

            submissions.append(
                Submission(
                    uco,
                    login,
                    path,
                    submission_id,
                    points,
                    hw_tag,
                    self.correction,
                    text,
                )
            )
            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():
            if not submissions[0].submitted_before_deadline:
                continue

            length = len(submissions)
            max_i = 0

            for i in range(1, length):
                if not submissions[i].submitted_before_deadline:
                    break

                if submissions[i].points >= submissions[max_i].points:
                    max_i = i

            submissions[max_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