PRO-450: Export rework, filter for projects by is_rated

1. Перерабока выгрузки:
Все оценки проектов на 1 листе xlsx файла.
Колонки:
ФИО|Email|Регион_РФ|Учебное_заведение|Название_учебного_заведения|Класс_курс|Фамилия эксперта|**criteria

2. Для проектов добавлен булевый фильтр `is_rated_by_expert`
Контекст использования `/program/{id}/projects` (страница "Проекты-участники") - для экспертов, чтобы сортировать была ли дана оценка проекту.
Сам фильтр на роуте: `/projects/`

partner_programs/admin.py

@@ -1,6 +1,6 @@
-import pandas as pd
 import tablib
 import re
+import urllib.parse
 from django.contrib import admin
 from django.db.models import QuerySet
 from django.http import HttpResponse, HttpRequest
@@ -11,7 +11,7 @@
 from partner_programs.models import PartnerProgram, PartnerProgramUserProfile
 from project_rates.models import Criteria, ProjectScore
 from projects.models import Project
-from users.models import Expert
+from partner_programs.services import XlsxFileToExport, ProjectScoreDataPreparer
 
 
 @admin.register(PartnerProgram)
@@ -154,86 +154,64 @@ def get_export_file(self, partner_program: PartnerProgram):
  return response
 
  def get_export_rates_view(self, request, object_id):
- criterias = Criteria.objects.filter(partner_program__id=object_id).select_related(
- "partner_program"
- )
- experts = Expert.objects.filter(programs=object_id).select_related("user")
- scores = ProjectScore.objects.filter(criteria__in=criterias)
- projects = Project.objects.filter(scores__in=scores)
- return self.get_export_rates(criterias, experts, scores, projects)
-
- def get_export_rates(self, criterias, experts, scores, projects):
- col_names = list(
- criterias.exclude(name="Комментарий").values_list("name", flat=True)
- )
-
- expert_names = [
- expert.user.first_name + " " + expert.user.last_name for expert in experts
- ]
-
- all_projects_data = []
- for project in projects:
- project_data = [[project.name, *col_names, "Комментарий"]]
-
- for expert, expert_name in zip(experts, expert_names):
- single_rate_data = [expert_name]
-
- scores_of_expert = []
- criterias_to_check = criterias.exclude(name="Комментарий")
- for criteria in criterias_to_check:
- checking_score = (
- scores.filter(
- criteria=criteria,
- user__first_name=expert.user.first_name,
- user__last_name=expert.user.last_name,
- project__name=project.name,
- )
- .exclude(criteria__name="Комментарий")
- .first()
- )
- if not checking_score:
- scores_of_expert.append("")
- else:
- scores_of_expert.append(checking_score.value)
-
- commentary = scores.filter(
- user__first_name=expert.user.first_name,
- user__last_name=expert.user.last_name,
- criteria__name="Комментарий",
- project__name=project.name,
- ).first()
- commentary = [commentary.value] if commentary else [""]
-
- scores_of_expert += commentary
-
- single_rate_data += scores_of_expert
+ rates_data_to_write: list[dict] = self._get_prepared_rates_data_for_export(object_id)
 
- project_data.append(single_rate_data)
- all_projects_data.append(project_data)
+ xlsx_file_writer = XlsxFileToExport()
+ xlsx_file_writer.write_data_to_xlsx(rates_data_to_write)
+ binary_data_to_export: bytes = xlsx_file_writer.get_binary_data_from_self_file()
+ xlsx_file_writer.delete_self_xlsx_file_from_local_machine()
 
- dataframed_projects_data = [
- pd.DataFrame(project_data) for project_data in all_projects_data
- ]
- with pd.ExcelWriter("output.xlsx") as writer:
- for df, pr_data in zip(dataframed_projects_data, all_projects_data):
- df.to_excel(writer, sheet_name=pr_data[0][0], index=False)
-
- with open("output.xlsx", "rb") as f:
- binary_data = f.read()
-
- # Формирование HTTP-ответа
- file_name = (
- f'{criterias.first().partner_program.name}_оценки {timezone.now().strftime("%d-%m-%Y %H:%M:%S")}'
+ encoded_file_name: str = urllib.parse.quote(
+ f'{PartnerProgram.objects.get(pk=object_id).name}_оценки {timezone.now().strftime("%d-%m-%Y %H:%M:%S")}'
  f".xlsx"
  )
  response = HttpResponse(
- binary_data,
+ binary_data_to_export,
  content_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
  )
- response["Content-Disposition"] = f'attachment; filename="{file_name}"'
-
+ response["Content-Disposition"] = f'attachment; filename*=UTF-8\'\'{encoded_file_name}'
  return response
 
+ def _get_prepared_rates_data_for_export(self, program_id: int) -> list[dict]:
+ """
+ Prepares info (list if dicts) for export about prjects_rates by experts.
+ Columns example:
+ ФИО|Email|Регион_РФ|Учебное_заведение|Название_учебного_заведения|Класс_курс|Фамилия эксперта|**criteria
+ """
+ criterias = Criteria.objects.filter(partner_program__id=program_id).select_related("partner_program")
+ scores = (
+ ProjectScore.objects
+ .filter(criteria__in=criterias)
+ .select_related("user", "criteria", "project")
+ .order_by("project", "criteria")
+ )
+ user_programm_profiles = (
+ PartnerProgramUserProfile.objects
+ .filter(partner_program__id=program_id)
+ .select_related("user")
+ )
+ projects = Project.objects.filter(scores__in=scores).select_related("leader").distinct()
+
+ # To reduce the number of DB requests.
+ user_profiles_dict: dict[int, PartnerProgramUserProfile] = {
+ profile.project_id: profile for profile in user_programm_profiles
+ }
+ scores_dict: dict[int, list[ProjectScore]] = {}
+ for score in scores:
+ scores_dict.setdefault(score.project_id, []).append(score)
+
+ prepared_projects_rates_data: list[dict] = []
+ for project in projects:
+ project_data_preparer = ProjectScoreDataPreparer(user_profiles_dict, scores_dict, project.id, program_id)
+ full_project_rates_data: dict = {
+ **project_data_preparer.get_project_user_info(),
+ **project_data_preparer.get_project_expert_info(),
+ **project_data_preparer.get_project_scores_info(),
+ }
+ prepared_projects_rates_data.append(full_project_rates_data)
+
+ return prepared_projects_rates_data
+
 
 @admin.register(PartnerProgramUserProfile)
 class PartnerProgramUserProfileAdmin(admin.ModelAdmin):

partner_programs/services.py

@@ -0,0 +1,128 @@
+import logging
+import os
+import pandas as pd
+
+from partner_programs.models import PartnerProgramUserProfile
+from project_rates.models import Criteria, ProjectScore
+
+
+logger = logging.getLogger()
+
+
+class XlsxFileToExport:
+ """
+ Writing data to `xlsx` file.
+ `filename` must contain `.xlsx` format prefix.
+ All data on 1 page.
+ """
+
+ def __init__(self, filename="output.xlsx"):
+ self.filename = filename
+
+ def write_data_to_xlsx(self, data: list[dict], sheet_name="scores") -> None:
+ try:
+ data_frames = pd.DataFrame(data)
+ with pd.ExcelWriter(self.filename) as writer:
+ data_frames.to_excel(writer, sheet_name=sheet_name, index=False)
+ except Exception as e:
+ logger.error(f"Write export rates data error: {str(e)}", exc_info=True)
+ raise
+
+ def get_binary_data_from_self_file(self) -> bytes:
+ try:
+ with open(self.filename, "rb") as f:
+ binary_data = f.read()
+ return binary_data
+ except Exception as e:
+ logger.error(f"Read export rates data error: {str(e)}", exc_info=True)
+ raise
+
+ def delete_self_xlsx_file_from_local_machine(self) -> None:
+ if os.path.isfile(self.filename) and self.filename.endswith(".xlsx"):
+ os.remove(self.filename)
+
+
+class ProjectScoreDataPreparer:
+ """
+ Data preparer about project_rates by experts.
+ """
+
+ USER_ERROR_FIELDS = {
+ "Фамилия": "ОШИБКА",
+ "Имя": "ОШИБКА",
+ "Отчество": "ОШИБКА",
+ "Email": "ОШИБКА",
+ "Регион_РФ": "ОШИБКА",
+ "Учебное_заведение": "ОШИБКА",
+ "Название_учебного_заведения": "ОШИБКА",
+ "Класс_курс": "ОШИБКА",
+ }
+
+ EXPERT_ERROR_FIELDS = {
+ "Фамилия эксперта": "ОШИБКА"
+ }
+
+ def __init__(
+ self,
+ user_profiles: dict[int, PartnerProgramUserProfile],
+ scores: dict[int, list[ProjectScore]],
+ project_id: int,
+ program_id: int
+ ):
+ self._project_id = project_id
+ self._user_profiles = user_profiles
+ self._scores = scores
+ self._program_id = program_id
+
+ def get_project_user_info(self) -> dict[str, str]:
+ try:
+ user_program_profile: PartnerProgramUserProfile = self._user_profiles.get(self._project_id)
+ user_program_profile_json: dict = user_program_profile.partner_program_data if user_program_profile else {}
+
+ user_info: dict[str, str] = {
+ "Фамилия": user_program_profile.user.last_name if user_program_profile else '',
+ "Имя": user_program_profile.user.first_name if user_program_profile else '',
+ "Отчество": user_program_profile.user.patronymic if user_program_profile else '',
+ "Email": (
+ user_program_profile_json.get('email') if user_program_profile_json.get('email')
+ else user_program_profile.user.email
+ ),
+ "Регион_РФ": user_program_profile_json.get('region', ''),
+ "Учебное_заведение": user_program_profile_json.get('education_type', ''),
+ "Название_учебного_заведения": user_program_profile_json.get('institution_name', ''),
+ "Класс_курс": user_program_profile_json.get('class_course', ''),
+ }
+ return user_info
+ except Exception as e:
+ logger.error(f"Prepare export rates data about user error: {str(e)}", exc_info=True)
+ return self.USER_ERROR_FIELDS
+
+ def get_project_expert_info(self) -> dict[str, str]:
+ try:
+ project_scores: list[ProjectScore] = self._scores.get(self._project_id, [])
+ first_score = project_scores[0] if project_scores else None
+ expert_last_name: dict[str, str] = {"Фамилия эксперта": first_score.user.last_name if first_score else ''}
+ return expert_last_name
+ except Exception as e:
+ logger.error(f"Prepare export rates data about expert error: {str(e)}", exc_info=True)
+ return self.EXPERT_ERROR_FIELDS
+
+ def get_project_scores_info(self) -> dict[str, str]:
+ try:
+ project_scores_dict = {}
+ project_scores: list[ProjectScore] = self._scores.get(self._project_id, [])
+ score_info_with_out_comment: dict[str, str] = {
+ score.criteria.name: score.value
+ for score in project_scores if score.criteria.name != "Комментарий"
+ }
+ project_scores_dict.update(score_info_with_out_comment)
+ comment = next((score for score in project_scores if score.criteria.name == "Комментарий"), None)
+ if comment is not None:
+ project_scores_dict["Комментарий"] = comment.value
+ return project_scores_dict
+ except Exception as e:
+ logger.error(f"Prepare export rates data about project_scores error: {str(e)}", exc_info=True)
+ return {
+ criteria.name: "ОШИБКА"
+ for criteria in Criteria.objects.filter(partner_program__id=self._program_id)
+ }

projects/filters.py

@@ -4,6 +4,7 @@
 from users.models import Expert
 from partner_programs.models import PartnerProgram, PartnerProgramUserProfile
 from projects.models import Project
+from project_rates.models import ProjectScore
 
 
 class ProjectFilter(filters.FilterSet):
@@ -76,6 +77,12 @@ def filter_by_partner_program(self, queryset, name, value):
  except PartnerProgram.DoesNotExist:
  return Project.objects.none()
 
+ def filter_by_have_expert_rates(self, queryset, name, value):
+ rated_projects_ids = ProjectScore.objects.values_list("project_id", flat=True).distinct()
+ if value:
+ return queryset.filter(id__in=rated_projects_ids)
+ return queryset.exclude(id__in=rated_projects_ids)
+
  name__contains = filters.Filter(field_name="name", lookup_expr="contains")
  description__contains = filters.Filter(
  field_name="description", lookup_expr="contains"
@@ -103,6 +110,10 @@ def filter_by_partner_program(self, queryset, name, value):
  partner_program = filters.NumberFilter(
  field_name="partner_program", method="filter_by_partner_program"
  )
+ is_rated_by_expert = filters.BooleanFilter(
+ method="filter_by_have_expert_rates",
+ label=("is_rated_by_expert\n`1`/`true` rated projects\n`0`/`false` dosn't rated")
+ )
 
  class Meta:
  model = Project