grades.py

        from course.utils import get_session_grading_rule

        try:
            flow_desc = get_flow_desc(pctx.repo, pctx.course,
                    opportunity.flow_id, pctx.course_commit_sha)
        except ObjectDoesNotExist:
            flow_sessions_and_session_properties: \
                list[tuple[Any, SessionProperties]] | None = None
        else:
            flow_sessions_and_session_properties = []
            for session in flow_sessions:
                adjust_flow_session_page_data(
                        pctx.repo, session, pctx.course.identifier,
                        flow_desc, respect_preview=False)

                grading_rule = get_session_grading_rule(
                        session, flow_desc, now_datetime)

                session_properties = SessionProperties(
                        due=grading_rule.due,
                        grade_description=grading_rule.description)
                flow_sessions_and_session_properties.append(
                        (session, session_properties))

    else:
        flow_sessions_and_session_properties = None

    avg_grade_percentage, avg_grade_population = average_grade(opportunity)

    show_privileged_info = pctx.has_permission(pperm.view_gradebook)
    show_page_grades = (
            show_privileged_info
            or opportunity.page_scores_in_participant_gradebook)

    # {{{ filter out pre-public grade changes

    if (not show_privileged_info
            and opportunity.hide_superseded_grade_history_before is not None):
        grade_changes = [gchange
                for gchange in grade_changes
                if not gchange.is_superseded
                or gchange.grade_time
                >= opportunity.hide_superseded_grade_history_before]

    # }}}

    return render_course_page(pctx, "course/gradebook-single.html", {
        "opportunity": opportunity,
        "avg_grade_percentage": avg_grade_percentage,
        "avg_grade_population": avg_grade_population,
        "grade_participation": participation,
        "grade_state_change_types": grade_state_change_types,
        "grade_changes": grade_changes,
        "state_machine": state_machine,
        "flow_sessions_and_session_properties": flow_sessions_and_session_properties,
        "show_privileged_info": show_privileged_info,
        "show_page_grades": show_page_grades,
        "allow_session_actions": (
            pperm.impose_flow_session_deadline
            or pperm.end_flow_session
            or pperm.regrade_flow_session
            or pperm.recalculate_flow_session_grade),
        })

# }}}


# {{{ import grades

class ImportGradesForm(StyledForm):
    def __init__(self, course, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.fields["grading_opportunity"] = forms.ModelChoiceField(
            queryset=(GradingOpportunity.objects
                .filter(course=course)
                .order_by("identifier")),
            help_text=_("Click to <a href='%s' target='_blank'>create</a> "
            "a new grading opportunity. Reload this form when done.")
            % reverse("admin:course_gradingopportunity_add"),
            label=pgettext_lazy("Field name in Import grades form",
                                "Grading opportunity"))

        self.fields["attempt_id"] = forms.CharField(
                initial="main",
                required=True,
                label=_("Attempt ID"))
        self.fields["file"] = forms.FileField(
                label=_("File"))

        self.fields["format"] = forms.ChoiceField(
                choices=(
                    ("csvhead", _("CSV with Header")),
                    ("csv", "CSV"),
                    ),
                label=_("Format"))

        self.fields["attr_type"] = forms.ChoiceField(
                choices=(
                    ("email_or_id", _("Email or NetID")),
                    ("institutional_id", _("Institutional ID")),
                    ("username", _("Username")),
                    ),
                label=_("User attribute"))

        self.fields["attr_column"] = forms.IntegerField(
                # Translators: the following strings are for the format
                # informatioin for a CSV file to be imported.
                help_text=_("1-based column index for the user attribute "
                "selected above to locate student record"),
                min_value=1,
                label=_("User attribute column"))
        self.fields["points_column"] = forms.IntegerField(
                help_text=_("1-based column index for the (numerical) grade"),
                min_value=1,
                label=_("Points column"))
        self.fields["feedback_column"] = forms.IntegerField(
                help_text=_("1-based column index for further (textual) feedback"),
                min_value=1, required=False,
                label=_("Feedback column"))
        self.fields["max_points"] = forms.DecimalField(
                initial=100,
                # Translators: "Max point" refers to full credit in points.
                label=_("Max points"))

        self.helper.add_input(Submit("preview", _("Preview")))
        self.helper.add_input(Submit("import", _("Import")))

    def clean(self):
        data = super().clean()
        attempt_id = data.get("attempt_id")
        if attempt_id:
            attempt_id = attempt_id.strip()
            flow_session_specific_attempt_id_prefix = "flow-session-"
            if attempt_id.startswith(flow_session_specific_attempt_id_prefix):
                self.add_error("attempt_id",
                               _('"%s" as a prefix is not allowed')
                               % flow_session_specific_attempt_id_prefix)

        file_contents = data.get("file")
        if file_contents:
            column_idx_list = [
                data["attr_column"],
                data["points_column"],
                data["feedback_column"]
            ]
            has_header = data["format"] == "csvhead"
            header_count = 1 if has_header else 0

            import io

            from course.utils import csv_data_importable
            importable, err_msg = csv_data_importable(
                    io.StringIO(
                        file_contents.read().decode("utf-8", errors="replace")),
                    column_idx_list,
                    header_count)

            if not importable:
                self.add_error("file", err_msg)


class ParticipantNotFound(ValueError):
    pass


def find_participant_from_user_attr(course, attr_type, attr_str):
    attr_str = attr_str.strip()

    exact_mode = "exact"
    if attr_type == "institutional_id":
        exact_mode = "iexact"
    kwargs = {f"user__{attr_type}__{exact_mode}": attr_str}

    matches = (Participation.objects
            .filter(
                course=course,
                status=participation_status.active,
                **kwargs)
            .select_related("user"))

    matches_count = matches.count()
    if not matches_count or matches_count > 1:
        from django.contrib.auth import get_user_model
        from django.utils.encoding import force_str
        attr_verbose_name = force_str(
            get_user_model()._meta.get_field(attr_type).verbose_name)

        map_dict = {"user_attr": attr_verbose_name, "user_attr_str": attr_str}

        if not matches_count:
            raise ParticipantNotFound(
                    _("no participant found with %(user_attr)s "
                    "'%(user_attr_str)s'") % map_dict)
        raise ParticipantNotFound(
                _("more than one participant found with %(user_attr)s "
                "'%(user_attr_str)s'") % map_dict)

    return matches[0]


def find_participant_from_id(course, id_str):
    id_str = id_str.strip().lower()

    matches = (Participation.objects
            .filter(
                course=course,
                status=participation_status.active,
                user__email__istartswith=id_str)
            .select_related("user"))

    surviving_matches = []
    for match in matches:
        if match.user.email.lower() == id_str:
            surviving_matches.append(match)
            continue

        email = match.user.email.lower()
        at_index = email.index("@")
        assert at_index > 0
        uid = email[:at_index]

        if uid == id_str:
            surviving_matches.append(match)
            continue

    if not surviving_matches:
        raise ParticipantNotFound(
                # Translators: use id_string to find user (participant).
                _("no participant found for '%(id_string)s'") % {
                    "id_string": id_str})
    if len(surviving_matches) > 1:
        raise ParticipantNotFound(
                _("more than one participant found for '%(id_string)s'") % {
                    "id_string": id_str})

    return surviving_matches[0]


def fix_decimal(s):
    if "," in s and "." not in s:
        comma_count = len([c for c in s if c == ","])
        if comma_count == 1:
            return s.replace(",", ".")
        else:
            return s

    else:
        return s


def points_equal(num: Decimal | None, other: Decimal | None) -> bool:
    if num is None and other is None:
        return True
    if ((num is None and other is not None)
            or (num is not None and other is None)):
        return False
    assert num is not None
    assert other is not None
    return abs(num - other) < 1e-2


def csv_to_grade_changes(
        log_lines,
        course, grading_opportunity, attempt_id, file_contents,
        attr_type, attr_column, points_column, feedback_column,
        max_points, creator, grade_time, has_header,

        # Row count limited to avoid out-of-memory situation.
        # https://github.com/inducer/relate/issues/849
        max_rows=10_000):
    result = []

    import csv

    from course.utils import get_col_contents_or_empty

    gchange_count = 0
    row_count = 0

    for row in csv.reader(file_contents):
        row_count += 1
        if row_count % 30 == 0:
            print(row_count)

        if row_count >= max_rows:
            raise ValueError(_(
                "Too many rows. "
                "Aborted processing after %d rows. "
                "Please split your file into smaller pieces.")
                % max_rows)

        if has_header:
            has_header = False
            continue

        gchange = GradeChange()
        gchange.opportunity = grading_opportunity
        try:
            if attr_type == "email_or_id":
                gchange.participation = find_participant_from_id(
                        course, get_col_contents_or_empty(row, attr_column-1))
            elif attr_type in ["institutional_id", "username"]:
                gchange.participation = find_participant_from_user_attr(
                        course, attr_type,
                        get_col_contents_or_empty(row, attr_column-1))
            else:
                raise NotImplementedError()
        except ParticipantNotFound as e:
            log_lines.append(e)
            continue

        gchange.state = grade_state_change_types.graded
        gchange.attempt_id = attempt_id

        points_str = get_col_contents_or_empty(row, points_column-1).strip()
        # Moodle's "NULL" grades look like this.
        if points_str in ["-", ""]:
            gchange.points = None
        else:
            gchange.points = Decimal(fix_decimal(points_str))

        gchange.max_points = max_points
        if feedback_column is not None:
            gchange.comment = get_col_contents_or_empty(row, feedback_column-1)

        gchange.creator = creator
        gchange.grade_time = grade_time

        last_grades = (GradeChange.objects
                .filter(
                    opportunity=grading_opportunity,
                    participation=gchange.participation,
                    attempt_id=gchange.attempt_id)
                .order_by("-grade_time")[:1])

        if last_grades.count():
            last_grade, = last_grades

            if last_grade.state == grade_state_change_types.graded:
                updated = []
                if not points_equal(last_grade.points, gchange.points):
                    updated.append(gettext("points"))
                if not points_equal(last_grade.max_points, gchange.max_points):
                    updated.append(gettext("max_points"))
                if last_grade.comment != gchange.comment:
                    updated.append(gettext("comment"))

                if updated:
                    log_lines.append(
                            string_concat(
                                "%(participation)s: %(updated)s ",
                                _("updated")
                                ) % {
                                    "participation": gchange.participation,
                                    "updated": ", ".join(updated)})

                    result.append(gchange)
            else:
                result.append(gchange)

        else:
            result.append(gchange)

        gchange_count += 1

    return gchange_count, result


@course_view
@transaction.atomic
def import_grades(pctx):
    if not pctx.has_permission(pperm.batch_import_grade):
        raise PermissionDenied(_("may not batch-import grades"))

    form_text = ""

    log_lines = []

    import io
    request = pctx.request
    if request.method == "POST":
        form = ImportGradesForm(
                pctx.course, request.POST, request.FILES)

        is_import = "import" in request.POST
        if form.is_valid():
            try:
                f = request.FILES["file"]
                f.seek(0)
                data = f.read().decode("utf-8", errors="replace")
                total_count, grade_changes = csv_to_grade_changes(
                        log_lines=log_lines,
                        course=pctx.course,
                        grading_opportunity=form.cleaned_data["grading_opportunity"],
                        attempt_id=form.cleaned_data["attempt_id"],
                        file_contents=io.StringIO(data),
                        attr_type=form.cleaned_data["attr_type"],
                        attr_column=form.cleaned_data["attr_column"],
                        points_column=form.cleaned_data["points_column"],
                        feedback_column=form.cleaned_data["feedback_column"],
                        max_points=form.cleaned_data["max_points"],
                        creator=request.user,
                        grade_time=now(),
                        has_header=form.cleaned_data["format"] == "csvhead")
            except Exception as e:
                messages.add_message(pctx.request, messages.ERROR,
                        string_concat(
                            pgettext_lazy("Starting of Error message",
                                "Error"),
                            ": %(err_type)s %(err_str)s")
                        % {
                            "err_type": type(e).__name__,
                            "err_str": str(e)})
            else:
                if total_count != len(grade_changes):
                    messages.add_message(pctx.request, messages.INFO,
                            _("%(total)d grades found, %(unchanged)d unchanged.")
                            % {"total": total_count,
                               "unchanged": total_count - len(grade_changes)})

                from django.template.loader import render_to_string

                if is_import:
                    GradeChange.objects.bulk_create(grade_changes)
                    form_text = render_to_string(
                            "course/grade-import-preview.html", {
                                "show_grade_changes": False,
                                "log_lines": log_lines,
                                })
                    messages.add_message(pctx.request, messages.SUCCESS,
                            _("%d grades imported.") % len(grade_changes))
                else:
                    form_text = render_to_string(
                            "course/grade-import-preview.html", {
                                "show_grade_changes": True,
                                "grade_changes": grade_changes,
                                "log_lines": log_lines,
                                })

    else:
        form = ImportGradesForm(pctx.course)

    return render_course_page(pctx, "course/generic-course-form.html", {
        "form_description": _("Import Grade Data"),
        "form": form,
        "form_text": form_text,
        })

# }}}


# {{{ download all submissions

class DownloadAllSubmissionsForm(StyledForm):
    def __init__(self, page_ids, session_tag_choices, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.fields["page_id"] = forms.ChoiceField(
                choices=tuple(
                    (pid, pid)
                    for pid in page_ids),
                label=_("Page ID"))
        self.fields["which_attempt"] = forms.ChoiceField(
                choices=(
                    ("first", _("Least recent attempt")),
                    ("last", _("Most recent attempt")),
                    ("all", _("All attempts")),
                    ),
                label=_("Attempts to include."),
                help_text=_(
                    "Every submission to the page counts as an attempt."),
                initial="last")
        self.fields["restrict_to_rules_tag"] = forms.ChoiceField(
                choices=session_tag_choices,
                help_text=_("Only download sessions tagged with this rules tag."),
                label=_("Restrict to rules tag"))
        self.fields["non_in_progress_only"] = forms.BooleanField(
                required=False,
                initial=True,
                help_text=_("Only download submissions from non-in-progress "
                    "sessions"),
                label=_("Non-in-progress only"))
        self.fields["include_feedback"] = forms.BooleanField(
                required=False,
                initial=False,
                help_text=_("Include provided feedback as text file in zip"),
                label=_("Include feedback"))
        self.fields["extra_file"] = forms.FileField(
                label=_("Additional File"),
                help_text=_(
                    "If given, the uploaded file will be included "
                    "in the zip archive. "
                    "If the produced archive is to be used for plagiarism "
                    "detection, then this may be used to include the reference "
                    "solution."),
                required=False)

        self.helper.add_input(
                Submit("download", _("Download")))


@course_view
def download_all_submissions(pctx, flow_id):
    if not pctx.has_permission(pperm.batch_download_submission):
        raise PermissionDenied(_("may not batch-download submissions"))

    from course.content import get_flow_desc
    flow_desc = get_flow_desc(pctx.repo, pctx.course, flow_id,
            pctx.course_commit_sha)

    # {{{ find access rules tag

    if hasattr(flow_desc, "rules"):
        access_rules_tags = getattr(flow_desc.rules, "tags", [])
    else:
        access_rules_tags = []

    ALL_SESSION_TAG = string_concat("<<<", _("ALL"), ">>>")  # noqa
    session_tag_choices = [
            (tag, tag)
            for tag in access_rules_tags] + [(ALL_SESSION_TAG,
                    string_concat("(", _("ALL"), ")"))]

    # }}}

    page_ids = [
            f"{group_desc.id}/{page_desc.id}"
            for group_desc in flow_desc.groups
            for page_desc in group_desc.pages]

    from course.page.base import AnswerFeedback

    request = pctx.request
    if request.method == "POST":
        form = DownloadAllSubmissionsForm(page_ids, session_tag_choices,
                request.POST)

        if form.is_valid():
            which_attempt = form.cleaned_data["which_attempt"]

            slash_index = form.cleaned_data["page_id"].index("/")
            group_id = form.cleaned_data["page_id"][:slash_index]
            page_id = form.cleaned_data["page_id"][slash_index+1:]

            from course.utils import PageInstanceCache
            page_cache = PageInstanceCache(pctx.repo, pctx.course, flow_id)

            visits = (FlowPageVisit.objects
                    .filter(
                        flow_session__course=pctx.course,
                        flow_session__flow_id=flow_id,
                        page_data__group_id=group_id,
                        page_data__page_id=page_id,
                        is_submitted_answer=True,
                        )
                    .select_related("flow_session")
                    .select_related("flow_session__participation__user")
                    .select_related("page_data")

                    # We overwrite earlier submissions with later ones
                    # in a dictionary below.
                    .order_by("visit_time"))

            if form.cleaned_data["non_in_progress_only"]:
                visits = visits.filter(flow_session__in_progress=False)

            if form.cleaned_data["restrict_to_rules_tag"] != ALL_SESSION_TAG:
                visits = (visits
                        .filter(
                            flow_session__access_rules_tag=(
                                form.cleaned_data["restrict_to_rules_tag"])))

            submissions = {}

            for visit in visits:
                page = page_cache.get_page(group_id, page_id,
                        pctx.course_commit_sha)

                from course.page import PageContext
                grading_page_context = PageContext(
                        course=pctx.course,
                        repo=pctx.repo,
                        commit_sha=pctx.course_commit_sha,
                        flow_session=visit.flow_session)

                bytes_answer = page.normalized_bytes_answer(
                        grading_page_context, visit.page_data.data,
                        visit.answer)

                if which_attempt in ["first", "last"]:
                    key = (visit.flow_session.participation.user.username,)
                elif which_attempt == "all":
                    key = (visit.flow_session.participation.user.username,
                            str(visit.flow_session.id))
                else:
                    raise NotImplementedError()

                if bytes_answer is not None:
                    if (which_attempt == "first"
                            and key in submissions):
                        # Already there, disregard further ones
                        continue

                    submissions[key] = (
                            bytes_answer, list(visit.grades.all()))

            from io import BytesIO
            from zipfile import ZipFile
            bio = BytesIO()
            with ZipFile(bio, "w") as subm_zip:
                for key, ((extension, bytes_answer), visit_grades) in \
                        submissions.items():
                    basename = "-".join(key)
                    subm_zip.writestr(
                            basename + extension,
                            bytes_answer)

                    if form.cleaned_data["include_feedback"]:
                        feedback_lines = []

                        feedback_lines.append(
                            "scores: {}".format(", ".join(
                                    str(g.correctness)
                                    for g in visit_grades)))

                        for i, grade in enumerate(visit_grades):
                            feedback_lines.append(75*"-")
                            feedback_lines.append(
                                "grade %i: score: %s" % (i+1, grade.correctness))
                            afb = AnswerFeedback.from_json(grade.feedback, None)
                            if afb is not None:
                                feedback_lines.append(afb.feedback)

                        subm_zip.writestr(
                                basename + "-feedback.txt",
                                "\n".join(feedback_lines))

                extra_file = request.FILES.get("extra_file")
                if extra_file is not None:
                    subm_zip.writestr(extra_file.name, extra_file.read())

            response = http.HttpResponse(
                    bio.getvalue(),
                    content_type="application/zip")
            response["Content-Disposition"] = (
                    f'attachment; filename="submissions_{pctx.course.identifier}_'
                    f'{flow_id}_{group_id}_{page_id}_'
                    f'{now().date().strftime("%Y-%m-%d")}.zip"')
            return response

    else:
        form = DownloadAllSubmissionsForm(page_ids, session_tag_choices)

    return render_course_page(pctx, "course/generic-course-form.html", {
        "form": form,
        "form_description": _("Download All Submissions in Zip file")
        })

# }}}


# {{{ edit_grading_opportunity

class EditGradingOpportunityForm(StyledModelForm):
    def __init__(self, add_new: bool, *args: Any, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)

        if not add_new:
            self.fields["identifier"].disabled = True

        self.fields["course"].disabled = True
        self.fields["flow_id"].disabled = True
        self.fields["creation_time"].disabled = True

        self.helper.add_input(
                Submit("submit", _("Update")))

    class Meta:
        model = GradingOpportunity
        exclude = (
                # do not exclude 'course', used in unique_together checking
                # not used
                "due_time",
                )
        widgets = {
                "hide_superseded_grade_history_before": HTML5DateTimeInput(),
                }


@course_view
def edit_grading_opportunity(pctx: CoursePageContext,
        opportunity_id: int) -> http.HttpResponse:
    if not pctx.has_permission(pperm.edit_grading_opportunity):
        raise PermissionDenied()

    request = pctx.request

    num_opportunity_id = int(opportunity_id)
    if num_opportunity_id == -1:
        gopp = GradingOpportunity(course=pctx.course)
        add_new = True
    else:
        gopp = get_object_or_404(GradingOpportunity, id=num_opportunity_id)
        add_new = False

    if gopp.course.id != pctx.course.id:
        raise SuspiciousOperation(
                "may not edit grading opportunity in different course")

    if request.method == "POST":
        form = EditGradingOpportunityForm(add_new, request.POST, instance=gopp)

        if form.is_valid():
            form.save()
            return redirect("relate-edit_grading_opportunity",
                    pctx.course.identifier, form.instance.id)

    else:
        form = EditGradingOpportunityForm(add_new, instance=gopp)

    return render_course_page(pctx, "course/generic-course-form.html", {
        "form_description": _("Edit Grading Opportunity"),
        "form": form
        })

# }}}

# vim: foldmethod=marker