Results: organizer ranking by case and Excel calculations

partner_programs/results_views.py

@@ -0,0 +1,36 @@
+from django.http import HttpResponse
+from rest_framework.response import Response
+
+from partner_programs.management_views import ManagementBase
+from partner_programs.services.management import managed_program
+from partner_programs.services.results import ResultsQuery, results_data
+from partner_programs.services.results_excel import export_results
+
+
+class ManagementResultsView(ManagementBase):
+    def get(self, request, pk):
+        program = managed_program(pk, request.user)
+        query = ResultsQuery(data=request.query_params)
+        query.is_valid(raise_exception=True)
+        values = dict(query.validated_data)
+        values["scope"] = "current"
+        response = Response(results_data(program, values))
+        response["Cache-Control"] = "private, no-store"
+        return response
+
+
+class ManagementResultsExportView(ManagementBase):
+    def get(self, request, pk):
+        program = managed_program(pk, request.user)
+        query = ResultsQuery(data=request.query_params)
+        query.is_valid(raise_exception=True)
+        data = results_data(program, query.validated_data)
+        response = HttpResponse(
+            export_results(data),
+            content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        )
+        response[
+            "Content-Disposition"
+        ] = f'attachment; filename="program-{pk}-results.xlsx"'
+        response["Cache-Control"] = "private, no-store"
+        return response

partner_programs/services/results.py

@@ -0,0 +1,192 @@
+"""Organizer results: one calculation for the ranking and its Excel export."""
+
+from collections import defaultdict
+from decimal import Decimal, ROUND_HALF_UP
+
+from django.utils import timezone
+from rest_framework import serializers
+
+from partner_programs.services.case_analytics import project_case_links
+from partner_programs.services.case_fields import get_program_case_field
+from partner_programs.services.management import person_name
+from project_rates.models import Criteria, ProjectExpertAssignment, ProjectScore
+from project_rates.validators import ProjectScoreValidator
+
+
+class ResultsQuery(serializers.Serializer):
+    search = serializers.CharField(
+        required=False, default="", allow_blank=True, max_length=200
+    )
+    case_scope = serializers.ChoiceField(
+        choices=["all", "selected", "without_case"], default="all"
+    )
+    case_name = serializers.CharField(required=False, allow_blank=True, max_length=500)
+    order = serializers.ChoiceField(choices=["asc", "desc"], default="desc")
+    scope = serializers.ChoiceField(choices=["current", "all"], default="current")
+
+
+def rounded(value):
+    return value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
+
+
+def numeric(value):
+    return float(value) if value is not None else None
+
+
+def valid_score(criterion, value):
+    try:
+        ProjectScoreValidator.validate(
+            criteria_type=criterion.type,
+            value=value,
+            criteria_min_value=criterion.min_value,
+            criteria_max_value=criterion.max_value,
+        )
+        return True
+    except (TypeError, ValueError):
+        return False
+
+
+def results_data(program, query):
+    criteria = list(Criteria.objects.filter(partner_program=program).order_by("pk"))
+    numeric_criteria = [c for c in criteria if c.type in ("int", "float")]
+    field = get_program_case_field(program)
+    options = field.get_options_list() if field else []
+    if query["case_scope"] == "selected" and query.get("case_name") not in options:
+        raise serializers.ValidationError({"case_name": "Выберите существующий кейс."})
+    links = list(
+        project_case_links(program, field=field)
+        .filter(submitted=True)
+        .select_related("project")
+        .order_by("pk")
+    )
+    project_ids = [link.project_id for link in links]
+    values, people = defaultdict(dict), defaultdict(dict)
+    for score in ProjectScore.objects.filter(
+        criteria__partner_program=program, project_id__in=project_ids
+    ).select_related("user"):
+        values[(score.project_id, score.user_id)][score.criteria_id] = score.value
+        people[score.project_id][score.user_id] = score.user
+    for assignment in ProjectExpertAssignment.objects.filter(
+        partner_program=program, project_id__in=project_ids
+    ).select_related("expert__user"):
+        people[assignment.project_id][assignment.expert.user_id] = assignment.expert.user
+    required = program.required_evaluators or program.max_project_rates or 1
+    maximum = (
+        sum(Decimal(str(c.max_value)) for c in numeric_criteria)
+        if numeric_criteria and all(c.max_value is not None for c in numeric_criteria)
+        else None
+    )
+    buckets = {name: [] for name in [*options, None]}
+    for link in links:
+        experts = []
+        for user_id, user in sorted(people[link.project_id].items()):
+            scores = values[(link.project_id, user_id)]
+            complete = bool(criteria) and all(
+                c.pk in scores and valid_score(c, scores[c.pk]) for c in criteria
+            )
+            total = (
+                sum(Decimal(scores[c.pk]) for c in numeric_criteria)
+                if complete and numeric_criteria
+                else None
+            )
+            experts.append(
+                {
+                    "user_id": user_id,
+                    "name": person_name(user),
+                    "email": user.email,
+                    "avatar": user.avatar or None,
+                    "complete": complete,
+                    "total": numeric(total),
+                    "values": [
+                        {"criterion_id": c.pk, "value": scores.get(c.pk)}
+                        for c in criteria
+                    ],
+                }
+            )
+        completed = [e for e in experts if e["complete"]]
+        score = (
+            rounded(sum(Decimal(str(e["total"])) for e in completed) / len(completed))
+            if completed and numeric_criteria
+            else None
+        )
+        averages = {
+            str(c.pk): numeric(
+                rounded(
+                    sum(
+                        Decimal(values[(link.project_id, e["user_id"])][c.pk])
+                        for e in completed
+                    )
+                    / len(completed)
+                )
+            )
+            for c in numeric_criteria
+            if completed
+        }
+        buckets.setdefault(link.case_name, []).append(
+            {
+                "id": link.pk,
+                "project_id": link.project_id,
+                "name": link.project.name,
+                "case_name": link.case_name,
+                "completed": len(completed),
+                "required": required,
+                "ready": len(completed) >= required and score is not None,
+                "score": numeric(score),
+                "place": None,
+                "averages": averages,
+                "experts": experts,
+            }
+        )
+    groups = []
+    for case_name, projects in buckets.items():
+        ready = sorted(
+            [p for p in projects if p["ready"]],
+            key=lambda p: (-p["score"], p["name"].lower(), p["id"]),
+        )
+        last_score, place = None, 0
+        for index, project in enumerate(ready, 1):
+            if project["score"] != last_score:
+                place = index
+            project["place"] = place
+            last_score = project["score"]
+        if query["scope"] != "all":
+            if query["case_scope"] == "selected" and case_name != query.get("case_name"):
+                continue
+            if query["case_scope"] == "without_case" and case_name is not None:
+                continue
+            projects = [
+                p for p in projects if query["search"].lower() in p["name"].lower()
+            ]
+        sign = -1 if query["order"] == "desc" else 1
+        projects.sort(
+            key=lambda p: (
+                not p["ready"],
+                p["score"] is None,
+                sign * (p["score"] or 0),
+                p["name"].lower(),
+                p["id"],
+            )
+        )
+        if projects:
+            groups.append({"case_name": case_name, "projects": projects})
+    return {
+        "program_id": program.pk,
+        "program_name": program.name,
+        "generated_at": timezone.now(),
+        "required_evaluators": required,
+        "maximum": numeric(maximum),
+        "has_numeric_criteria": bool(numeric_criteria),
+        "case_options": options,
+        "groups": groups,
+        "criteria": [
+            {
+                "id": c.pk,
+                "name": c.name,
+                "description": c.description,
+                "type": c.type,
+                "min_value": c.min_value,
+                "max_value": c.max_value,
+            }
+            for c in criteria
+        ],
+    }

partner_programs/services/results_excel.py

@@ -0,0 +1,143 @@
+"""Typed cells and real formulas, from the same results snapshot as the UI."""
+
+from io import BytesIO
+from math import isfinite
+
+from openpyxl import Workbook
+from openpyxl.styles import Font, PatternFill
+from openpyxl.utils import get_column_letter
+from openpyxl.workbook.properties import CalcProperties
+from openpyxl.cell.cell import ILLEGAL_CHARACTERS_RE
+
+
+def text_cell(sheet, row, column, value):
+    cell = sheet.cell(
+        row,
+        column,
+        ILLEGAL_CHARACTERS_RE.sub("", str(value)) if value is not None else "",
+    )
+    cell.data_type = "s"  # User strings must never become executable Excel formulas.
+
+
+def export_results(data):
+    workbook = Workbook()
+    workbook.calculation = CalcProperties(fullCalcOnLoad=True)
+    rating = workbook.active
+    rating.title = "Рейтинг"
+    experts = workbook.create_sheet("Оценки экспертов")
+    criteria = data["criteria"]
+    numeric = [c for c in criteria if c["type"] in ("int", "float")]
+    other = [c for c in criteria if c not in numeric]
+    rating.append(
+        [
+            "Место",
+            "Проект",
+            "Кейс",
+            "Получено оценок",
+            "Требуется оценок",
+            *[c["name"] + " (среднее)" for c in numeric],
+            "Итог",
+            "Статус",
+            "ID проекта",
+        ]
+    )
+    experts.append(
+        [
+            "Проект",
+            "Кейс",
+            "Эксперт",
+            "Email",
+            "Статус",
+            *[c["name"] for c in numeric],
+            "Сумма",
+            *[c["name"] for c in other],
+            "ID проекта",
+            "ID эксперта",
+        ]
+    )
+    total_col = 6 + len(numeric)
+    for group in data["groups"]:
+        for project in group["projects"]:
+            expert_rows = []
+            start_row = experts.max_row + 1
+            for expert in project["experts"]:
+                row = experts.max_row + 1
+                for col, value in enumerate(
+                    [
+                        project["name"],
+                        project["case_name"],
+                        expert["name"],
+                        expert["email"],
+                        "Полная" if expert["complete"] else "Неполная",
+                    ],
+                    1,
+                ):
+                    text_cell(experts, row, col, value)
+                values = {v["criterion_id"]: v["value"] for v in expert["values"]}
+                for col, criterion in enumerate(numeric, 6):
+                    value = values.get(criterion["id"])
+                    if value is not None and value != "":
+                        try:
+                            number = float(value)
+                            if not isfinite(number):
+                                raise ValueError()
+                            experts.cell(row, col, number)
+                        except (TypeError, ValueError):
+                            text_cell(experts, row, col, value)
+                if expert["complete"]:
+                    expert_rows.append(row)
+                    if numeric:
+                        experts.cell(
+                            row,
+                            total_col,
+                            f"=SUM(F{row}:{get_column_letter(total_col - 1)}{row})",
+                        )
+                for col, criterion in enumerate(other, total_col + 1):
+                    value = values.get(criterion["id"])
+                    if criterion["type"] == "bool" and value in ("True", "False"):
+                        value = "Да" if value == "True" else "Нет"
+                    text_cell(experts, row, col, value)
+                experts.cell(row, total_col + len(other) + 1, project["project_id"])
+                experts.cell(row, total_col + len(other) + 2, expert["user_id"])
+            row = rating.max_row + 1
+            if project["place"] is not None:
+                rating.cell(row, 1, project["place"])
+            text_cell(rating, row, 2, project["name"])
+            text_cell(rating, row, 3, project["case_name"] or "Без кейса")
+            rating.cell(row, 4, project["completed"])
+            rating.cell(row, 5, project["required"])
+            for index in range(len(numeric) + 1):
+                if not expert_rows or not numeric:
+                    continue
+                col = 6 + index
+                last_row = experts.max_row
+                statuses = f"'Оценки экспертов'!$E${start_row}:$E${last_row}"
+                column = get_column_letter(col)
+                refs = f"'Оценки экспертов'!{column}{start_row}:{column}{last_row}"
+                formula = f'AVERAGEIF({statuses},"Полная",{refs})'
+                if index == len(numeric):
+                    formula = f"ROUND({formula},2)"
+                rating.cell(row, col, "=" + formula).number_format = "0.00"
+            status = (
+                "Нет балльных критериев"
+                if not numeric
+                else "В рейтинге"
+                if project["ready"]
+                else "Предварительно"
+                if project["completed"]
+                else "Нет оценок"
+            )
+            text_cell(rating, row, total_col + 1, status)
+            rating.cell(row, total_col + 2, project["project_id"])
+    for sheet in (rating, experts):
+        sheet.freeze_panes = "D2"
+        sheet.auto_filter.ref = sheet.dimensions
+        for cell in sheet[1]:
+            cell.data_type = "s"
+            cell.font = Font(bold=True, color="FFFFFF")
+            cell.fill = PatternFill("solid", fgColor="8A63E6")
+        for col in range(1, sheet.max_column + 1):
+            sheet.column_dimensions[get_column_letter(col)].width = 24
+    output = BytesIO()
+    workbook.save(output)
+    return output.getvalue()