mamweb/vysledkovky/utils.py

import abc
from functools import cached_property
from typing import Union, Iterable  # TODO: s pythonem 3.10 přepsat na '|'

from tvorba.models import Rocnik, Cislo, Deadline, Problem, Clanek
from odevzdavatko.models import Hodnoceni
from personalni.models import Resitel
from soustredeni.models import Konfera
from django.db.models import Q, Sum
from seminar.utils import resi_v_rocniku

ROCNIK_ZRUSENI_TEMAT = 25


class FixedIterator:
	def next(self):
		return self.niter.__next__()

	def __init__(self, niter):
		self.niter = niter


def body_resitelu(
		za: Union[Cislo, Rocnik, None] = None,
		do: Deadline = None,
		od: Deadline = None,
		jen_verejne: bool = True,
		resitele: Iterable[Resitel] = None,
		null=0 # Výchozí hodnota, pokud pro daného řešitele nejsou body
) -> dict[int, int]:
	filtr = Q()

	if jen_verejne:
		filtr &= Q(reseni__hodnoceni__deadline_body__verejna_vysledkovka=True)

	# Zjistíme, typ objektu v parametru "za"
	if isinstance(za, Rocnik):
		filtr &= Q(reseni__hodnoceni__deadline_body__cislo__rocnik=za)
	elif isinstance(za, Cislo):
		filtr &= Q(reseni__hodnoceni__deadline_body__cislo=za)

	if do:
		filtr &= Q(reseni__hodnoceni__deadline_body__deadline__lte=do.deadline)

	if od:
		filtr &= Q(reseni__hodnoceni__deadline_body__deadline__gte=od.deadline)

	resiteleQuery = Resitel.objects.all()

	if resitele is not None:
		resitele_id = [r.id for r in resitele]
		resiteleQuery = resiteleQuery.filter(id__in=resitele_id)

	# Přidáme ke každému řešiteli údaj ".body" se součtem jejich bodů
	resitele_s_body = resiteleQuery.annotate(
		body=Sum('reseni__hodnoceni__body', filter=filtr))

	# Teď jen z QuerySetu řešitelů anotovaných body vygenerujeme slovník
	# indexovaný řešitelským id obsahující body.
	# Pokud jsou body None, nahradíme za 0.
	slovnik = {
		int(res.id): (res.body if res.body else null) for res in resitele_s_body
	}
	return slovnik


class Vysledkovka(abc.ABC):
	jen_verejne: bool
	rocnik: Rocnik
	do_deadlinu: Deadline

	@property
	@abc.abstractmethod
	def aktivni_resitele(self) -> list[Resitel]:
		...

	@cached_property
	def resitele_s_body_za_rocnik_setrizeny_seznam(self) -> list[tuple[int, int]]:
		# spočítáme všem řešitelům jejich body za ročník
		resitel_body_za_rocnik_slovnik = body_resitelu(
			resitele=self.aktivni_resitele,
			za=self.rocnik,
			jen_verejne=self.jen_verejne,
			do=self.do_deadlinu
		)

		# zeptáme se na dvojice (řešitel, body) za ročník a setřídíme sestupně
		resitele_s_body_za_rocnik_setrizeny_seznam = sorted(
			resitel_body_za_rocnik_slovnik.items(),
			key=lambda x: x[1], reverse=True
		)

		return resitele_s_body_za_rocnik_setrizeny_seznam

	@cached_property
	def body_za_rocnik_seznamy(self) -> tuple[list[int], list[int]]:
		if len(self.resitele_s_body_za_rocnik_setrizeny_seznam) == 0:
			return [], []
		return tuple(zip(*self.resitele_s_body_za_rocnik_setrizeny_seznam))

	@property
	def setrizeni_resitele_id(self) -> list[int]:
		return self.body_za_rocnik_seznamy[0]

	@property
	def setrizene_body(self) -> list[int]:
		return self.body_za_rocnik_seznamy[1]

	@cached_property
	def resitel_body_odjakziva_slovnik(self) -> dict[int, int]:
		return body_resitelu(jen_verejne=self.jen_verejne, do=self.do_deadlinu)

	@cached_property
	def poradi(self):
		# ze seznamu obsahujícího setřízené body spočítáme sloupec s pořadím
		aktualni_poradi = 1
		sloupec_s_poradim = []

		# seskupíme seznam všech bodů podle hodnot
		for index in range(0, len(self.setrizene_body)):
			# pokud je pořadí větší než číslo řádku, tak jsme vypsali větší rozsah
			# a chceme vypsat už jen prázdné místo, než dojdeme na správný řádek
			if (index + 1) < aktualni_poradi:
				sloupec_s_poradim.append("")
				continue
			velikost_skupiny = 0
			# zjistíme počet po sobě jdoucích stejných hodnot
			while self.setrizene_body[index] == self.setrizene_body[
				index + velikost_skupiny]:
				velikost_skupiny += 1
				# na konci musíme ošetřit přetečení seznamu
				if (index + velikost_skupiny) > len(self.setrizene_body) - 1:
					break
			# pokud je velikost skupiny 1, vypíšu pořadí
			if velikost_skupiny == 1:
				sloupec_s_poradim.append(f"{aktualni_poradi}.")
			# pokud je skupina větší, vypíšu rozsah
			else:
				sloupec_s_poradim.append(
					f"{aktualni_poradi}.–{aktualni_poradi + velikost_skupiny - 1}."
				)
			# zvětšíme aktuální pořadí o tolik, kolik pozic bylo přeskočeno
			aktualni_poradi += velikost_skupiny
		return sloupec_s_poradim


class VysledkovkaRocniku(Vysledkovka):

	def __init__(self, rocnik: Rocnik, jen_verejne: bool = True):
		self.rocnik = rocnik
		self.jen_verejne = jen_verejne
		deadliny = Deadline.objects.filter(cislo__rocnik=rocnik)
		if jen_verejne:
			deadliny = deadliny.filter(verejna_vysledkovka=True)
		self.do_deadlinu = deadliny.order_by("deadline").last()

	@cached_property
	def aktivni_resitele(self) -> list[Resitel]:
		return list(resi_v_rocniku(self.rocnik))

	@cached_property
	def cisla_rocniku(self) -> list[Cislo]:
		""" Vrátí všechna čísla daného ročníku. """
		if self.jen_verejne:
			return self.rocnik.verejne_vysledkovky_cisla()
		else:
			return self.rocnik.cisla.all().order_by('poradi')

	@cached_property
	def body_za_cisla_slovnik(self) -> dict[int, dict[int, int]]: # Výstup: Cislo.id → ( Resitel.id → body )
		# TODO: Body jsou decimal!
		body_cisla_slovnik = dict()
		for cislo in self.cisla_rocniku:
			# získáme body za číslo
			body_za_cislo = body_resitelu(
				za=cislo,
				resitele=self.aktivni_resitele,
				jen_verejne=self.jen_verejne,
				null=""
			)
			body_cisla_slovnik[cislo.id] = body_za_cislo
		return body_cisla_slovnik

	class RadekVysledkovkyRocniku:
		# TODO: přepsat na dataclass
		""" Obsahuje věci, které se hodí vědět při konstruování výsledkovky.
		Umožňuje snazší práci v templatu (lepší, než seznam)."""

		def __init__(self, poradi, resitel, body_cisla_seznam, body_rocnik, body_odjakziva, rok):
			self.poradi = poradi
			self.resitel = resitel
			self.rocnik_resitele = resitel.rocnik(rok)
			self.body_rocnik = body_rocnik
			self.body_celkem_odjakziva = body_odjakziva
			self.body_cisla_seznam = body_cisla_seznam
			self.titul = resitel.get_titul(body_odjakziva)

	@cached_property
	def radky_vysledkovky(self) -> list[RadekVysledkovkyRocniku]:
		radky_vysledkovky = []

		setrizeni_resitele_dict = dict()
		for r in Resitel.objects.filter(
				id__in=self.setrizeni_resitele_id
		).select_related('osoba'):
			setrizeni_resitele_dict[r.id] = r

		for i, ar_id in enumerate(self.setrizeni_resitele_id):
			if self.setrizene_body[i] > 0:
				# seznam počtu bodů daného řešitele pro jednotlivá čísla
				body_cisla_seznam = []
				for cislo in self.cisla_rocniku:
					body_cisla_seznam.append(self.body_za_cisla_slovnik[cislo.id][ar_id])

				# Pokud řešitel dostal nějaké body
				if self.resitele_s_body_za_rocnik_setrizeny_seznam[i] != 0:
					# vytáhneme informace pro daného řešitele
					radek = self.RadekVysledkovkyRocniku(
						poradi=self.poradi[i],
						resitel=setrizeni_resitele_dict[ar_id],
						body_cisla_seznam=body_cisla_seznam,
						body_rocnik=self.setrizene_body[i],
						body_odjakziva=self.resitel_body_odjakziva_slovnik[ar_id],
						rok=self.rocnik)  # ročník semináře pro získání ročníku řešitele
					radky_vysledkovky.append(radek)

		return radky_vysledkovky


class VysledkovkaCisla(Vysledkovka):
	def __init__(
			self,
			cislo: Cislo,
			jen_verejne: bool = True,
			do_deadlinu: Deadline = None
	):
		self.cislo = cislo
		self.rocnik = cislo.rocnik
		self.jen_verejne = jen_verejne
		if do_deadlinu is None:
			do_deadlinu = Deadline.objects.filter(cislo=cislo).last()
		self.do_deadlinu = do_deadlinu

	@cached_property
	def aktivni_resitele(self) -> list[Resitel]:
		# TODO možná chytřeji vybírat aktivní řešitele
		return list(resi_v_rocniku(self.rocnik))

	@cached_property
	def problemy(self) -> list[Problem]:
		""" Vrátí seznam všech problémů s body v daném čísle. """
		return Problem.objects.filter(
			hodnoceni__in=Hodnoceni.objects.filter(deadline_body__cislo=self.cislo)
		).distinct().non_polymorphic().select_related('nadproblem').select_related('nadproblem__nadproblem')

	@cached_property
	def hlavni_problemy(self) -> list[Problem]:
		""" Vrátí seznam všech problémů, které již nemají nadproblém. """
		# hlavní problémy čísla
		# (mají vlastní sloupeček ve výsledkovce, nemají nadproblém)
		hlavni_problemy = set()
		for p in self.problemy:
			hlavni_problemy.add(p.hlavni_problem)

		# zunikátnění
		hlavni_problemy = list(hlavni_problemy)
		hlavni_problemy.sort(
			key=lambda k: k.kod_v_rocniku)  # setřídit podle t1, t2, c3, ...

		return hlavni_problemy

	# Není cached, protože si myslím, že queryset lze použít ve for jen jednou.
	@property
	def hodnoceni_do_cisla(self):
		hodnoceni = Hodnoceni.objects.prefetch_related('reseni__resitele').select_related('problem', 'reseni')
		if self.jen_verejne:
			hodnoceni = hodnoceni.filter(deadline_body__verejna_vysledkovka=True)
		return hodnoceni.filter(
			deadline_body__cislo=self.cislo,
			deadline_body__deadline__lte=self.do_deadlinu.deadline,
			body__isnull=False,
		)

	@cached_property
	def sectene_body(self):
		"""
			Sečte body za číslo, hlavní problémy a podproblémy.

			Problém s ID '-1' znamená problémy bez nadproblémů, jež nejsou témata, tj. články, úlohy, konfery, …
		"""

		# Body za číslo
		body_za_cislo = {ar.id: "" for ar in self.aktivni_resitele}

		# Body za hlavní problémy
		body_za_temata = {
			hp.id: {ar.id: "" for ar in self.aktivni_resitele}
			for hp in self.temata_a_spol
		}
		# Ostatní body
		body_za_temata[-1] = {ar.id: "" for ar in self.aktivni_resitele}

		# Body za podproblémy
		body_za_problemy = {
			tema.id: {
				problem.id: {ar.id: "" for ar in self.aktivni_resitele}
				for problem in self.podproblemy[tema.id]
			}
			for tema in self.temata_a_spol
		}
		# Ostatní body
		body_za_problemy[-1] = {
				problem.id: {ar.id: "" for ar in self.aktivni_resitele}
				for problem in self.podproblemy[-1]
		}

		# Sečteme hodnocení
		for hodnoceni in self.hodnoceni_do_cisla:
			prob = hodnoceni.problem
			nadproblem = prob.hlavni_problem.id

			# Když nadproblém není "téma", pak je "Ostatní"
			if nadproblem not in body_za_temata:
				nadproblem = -1

			problem_slovnik = body_za_problemy[nadproblem][prob.id]
			nadproblem_slovnik = body_za_temata[nadproblem]

			body = hodnoceni.body

			# Může mít více řešitelů
			for resitel in hodnoceni.reseni.resitele.all():
				if resitel not in self.aktivni_resitele:
					continue
				self.pricti_body(body_za_cislo, resitel, body)
				self.pricti_body(nadproblem_slovnik, resitel, body)
				self.pricti_body(problem_slovnik, resitel, body)
		return body_za_cislo, body_za_temata, body_za_problemy

	@cached_property
	def body_za_temata(self) -> dict[int, dict[int, str]]:
		return self.sectene_body[1]

	@cached_property
	def body_za_cislo(self) -> dict[int, str]:
		return self.sectene_body[0]

	@cached_property
	def problemy_slovnik(self):
		return self.sectene_body[2]

	@cached_property
	def temata_a_spol(self) -> list[Problem]:
		if self.rocnik.rocnik < ROCNIK_ZRUSENI_TEMAT:
			return self.hlavni_problemy
		else:
			return list(filter(self.ne_clanek_ne_konfera, self.hlavni_problemy))

	@cached_property
	def je_nejake_ostatni(self):
		return len(self.hlavni_problemy) - len(self.temata_a_spol) > 0

	@cached_property
	def podproblemy(self) -> dict[int, list[Problem]]:
		podproblemy = {hp.id: [] for hp in self.temata_a_spol}
		temata_a_spol = set(self.temata_a_spol)
		podproblemy[-1] = []

		for problem in self.problemy:
			h_problem = problem.hlavni_problem
			if h_problem in temata_a_spol:
				podproblemy[h_problem.id].append(problem)
			else:
				podproblemy[-1].append(problem)

		for podproblem in podproblemy.keys():
			def int_or_zero(p):
				try:
					return int(p.kod)
				except ValueError:
					return 0

			podproblemy[podproblem] = sorted(podproblemy[podproblem], key=int_or_zero)
		return podproblemy

	@cached_property
	def podproblemy_seznam(self) -> list[list[Problem]]:
		return [self.podproblemy[it.id] for it in self.temata_a_spol] + [self.podproblemy[-1]]

	@cached_property
	def podproblemy_iter(self) -> FixedIterator:
		return FixedIterator(self.podproblemy_seznam.__iter__())

	class RadekVysledkovkyCisla(object):
		# TODO: Přepsat na dataclass
		"""Obsahuje věci, které se hodí vědět při konstruování výsledkovky.
		Umožňuje snazší práci v templatu (lepší, než seznam)."""

		def __init__(self, poradi, resitel, temata_seznamk, body_cislo, body_rocnik, body_odjakziva, rok, body_podproblemy, body_podproblemy_iter):
			self.resitel = resitel
			self.rocnik_resitele = resitel.rocnik(rok)
			self.body_cislo = body_cislo
			self.body_rocnik = body_rocnik
			self.body_celkem_odjakziva = body_odjakziva
			self.poradi = poradi
			self.body_za_temata_seznam = temata_seznamk
			self.titul = resitel.get_titul(body_odjakziva)
			self.body_podproblemy = body_podproblemy
			self.body_podproblemy_iter = body_podproblemy_iter

	@cached_property
	def radky_vysledkovky(self) -> list[RadekVysledkovkyCisla]:
		# vytvoříme jednotlivé sloupce výsledkovky
		radky_vysledkovky = []

		setrizeni_resitele_slovnik = {}
		setrizeni_resitele = Resitel.objects.filter(id__in=self.setrizeni_resitele_id).select_related('osoba')

		for r in setrizeni_resitele:
			setrizeni_resitele_slovnik[r.id] = r

		for i, ar_id in enumerate(self.setrizeni_resitele_id):
			if self.setrizene_body[i] > 0:
				# získáme seznam bodů za problémy pro daného řešitele
				body_problemy = []
				body_podproblemy = []
				for hp in self.temata_a_spol:
					body_problemy.append(self.body_za_temata[hp.id][ar_id])
					body_podproblemy.append([
						self.problemy_slovnik[hp.id][it.id][ar_id]
						for it in self.podproblemy[hp.id]
					])
				if self.je_nejake_ostatni:
					body_problemy.append(self.body_za_temata[-1][ar_id])
					body_podproblemy.append(
						[self.problemy_slovnik[-1][it.id][ar_id] for it in self.podproblemy[-1]])
				# vytáhneme informace pro daného řešitele
				radek = self.RadekVysledkovkyCisla(
					poradi=self.poradi[i],
					resitel=setrizeni_resitele_slovnik[ar_id],
					temata_seznamk=body_problemy,
					body_cislo=self.body_za_cislo[ar_id],
					body_rocnik=self.setrizene_body[i],
					body_odjakziva=self.resitel_body_odjakziva_slovnik[ar_id],
					rok=self.rocnik,
					body_podproblemy=body_podproblemy,  # body všech podproblémů
					body_podproblemy_iter=FixedIterator(body_podproblemy.__iter__())
				)  # ročník semináře pro zjištění ročníku řešitele
				radky_vysledkovky.append(radek)
		return radky_vysledkovky

	@staticmethod
	def pricti_body(slovnik, resitel, body):
		""" Přiřazuje danému řešiteli body do slovníku. """
		# testujeme na None (""), pokud je to první řešení
		# daného řešitele, předěláme na 0
		# (v dalším kroku přičteme reálný počet bodů),
		# rozlišujeme tím mezi 0 a neodevzdaným řešením

		if slovnik[resitel.id] == "":
			slovnik[resitel.id] = 0

		slovnik[resitel.id] += body

	@staticmethod
	def ne_clanek_ne_konfera(problem):
		inst = problem.get_real_instance()
		return not (isinstance(inst, Clanek) or isinstance(inst, Konfera))


class VysledkovkaDoTeXu(VysledkovkaCisla):
	def __init__(
			self,
			nejake_cislo: Cislo,
			od_vyjma: Deadline,
			do_vcetne: Deadline
	):
		super().__init__(nejake_cislo, False, do_vcetne)
		self.od_deadlinu = od_vyjma

	@cached_property
	def problemy(self) -> list[Problem]:
		return Problem.objects.filter(hodnoceni__in=Hodnoceni.objects.filter(
				deadline_body__deadline__gt=self.od_deadlinu.deadline,
				deadline_body__deadline__lte=self.do_deadlinu.deadline,
			)).distinct().non_polymorphic().select_related('nadproblem').select_related('nadproblem__nadproblem')

	@property
	def hodnoceni_do_cisla(self):
		hodnoceni = Hodnoceni.objects.prefetch_related(
			'problem', 'reseni', 'reseni__resitele')
		if self.jen_verejne:
			hodnoceni = hodnoceni.filter(deadline_body__verejna_vysledkovka=True)
		return hodnoceni.filter(
			deadline_body__deadline__gt=self.od_deadlinu.deadline,
			deadline_body__deadline__lte=self.do_deadlinu.deadline,
			body__isnull=False,
		)