274 lines
11 KiB
Python
274 lines
11 KiB
Python
# Copyright (C) 2024 by Animath
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
from asgiref.sync import sync_to_async
|
|
from django.contrib.auth.models import User
|
|
from django.db import models
|
|
from django.db.models import Q, QuerySet
|
|
from django.utils.text import format_lazy
|
|
from django.utils.translation import gettext_lazy as _
|
|
from participation.models import Pool, Team, Tournament
|
|
from registration.models import ParticipantRegistration, Registration, VolunteerRegistration
|
|
from tfjm.permissions import PermissionType
|
|
|
|
|
|
class Channel(models.Model):
|
|
class ChannelCategory(models.TextChoices):
|
|
GENERAL = 'general', _("General channels")
|
|
TOURNAMENT = 'tournament', _("Tournament channels")
|
|
TEAM = 'team', _("Team channels")
|
|
PRIVATE = 'private', _("Private channels")
|
|
|
|
name = models.CharField(
|
|
max_length=255,
|
|
verbose_name=_("name"),
|
|
)
|
|
|
|
category = models.CharField(
|
|
max_length=255,
|
|
verbose_name=_("category"),
|
|
choices=ChannelCategory,
|
|
default=ChannelCategory.GENERAL,
|
|
)
|
|
|
|
read_access = models.CharField(
|
|
max_length=16,
|
|
verbose_name=_("read permission"),
|
|
choices=PermissionType,
|
|
)
|
|
|
|
write_access = models.CharField(
|
|
max_length=16,
|
|
verbose_name=_("write permission"),
|
|
choices=PermissionType,
|
|
)
|
|
|
|
tournament = models.ForeignKey(
|
|
'participation.Tournament',
|
|
on_delete=models.CASCADE,
|
|
blank=True,
|
|
null=True,
|
|
default=None,
|
|
verbose_name=_("tournament"),
|
|
related_name='chat_channels',
|
|
help_text=_("For a permission that concerns a tournament, indicates what is the concerned tournament."),
|
|
)
|
|
|
|
pool = models.ForeignKey(
|
|
'participation.Pool',
|
|
on_delete=models.CASCADE,
|
|
blank=True,
|
|
null=True,
|
|
default=None,
|
|
verbose_name=_("pool"),
|
|
related_name='chat_channels',
|
|
help_text=_("For a permission that concerns a pool, indicates what is the concerned pool."),
|
|
)
|
|
|
|
team = models.ForeignKey(
|
|
'participation.Team',
|
|
on_delete=models.CASCADE,
|
|
blank=True,
|
|
null=True,
|
|
default=None,
|
|
verbose_name=_("team"),
|
|
related_name='chat_channels',
|
|
help_text=_("For a permission that concerns a team, indicates what is the concerned team."),
|
|
)
|
|
|
|
private = models.BooleanField(
|
|
verbose_name=_("private"),
|
|
default=False,
|
|
help_text=_("If checked, only users who have been explicitly added to the channel will be able to access it."),
|
|
)
|
|
|
|
invited = models.ManyToManyField(
|
|
'auth.User',
|
|
verbose_name=_("invited users"),
|
|
related_name='+',
|
|
blank=True,
|
|
help_text=_("Extra users who have been invited to the channel, "
|
|
"in addition to the permitted group of the channel."),
|
|
)
|
|
|
|
def get_visible_name(self, user: User) -> str:
|
|
if self.private:
|
|
users = [f"{u.first_name} {u.last_name}" for u in self.invited.all() if u != user] \
|
|
or [f"{user.first_name} {user.last_name}"]
|
|
return ", ".join(users)
|
|
return self.name
|
|
|
|
def __str__(self):
|
|
return str(format_lazy(_("Channel {name}"), name=self.name))
|
|
|
|
@staticmethod
|
|
async def get_accessible_channels(user: User, permission_type: str = 'read') -> QuerySet["Channel"]:
|
|
permission_type = 'write_access' if 'write' in permission_type.lower() else 'read_access'
|
|
|
|
qs = Channel.objects.none()
|
|
if user.is_anonymous:
|
|
return Channel.objects.filter(**{permission_type: PermissionType.ANONYMOUS})
|
|
|
|
qs |= Channel.objects.filter(**{permission_type: PermissionType.AUTHENTICATED})
|
|
registration = await Registration.objects.prefetch_related('user').aget(user_id=user.id)
|
|
|
|
if registration.is_admin:
|
|
return Channel.objects.prefetch_related('invited').exclude(~Q(invited=user) & Q(private=True)).all()
|
|
|
|
if registration.is_volunteer:
|
|
registration = await VolunteerRegistration.objects \
|
|
.prefetch_related('jury_in__tournament', 'organized_tournaments').aget(user_id=user.id)
|
|
|
|
qs |= Channel.objects.filter(**{permission_type: PermissionType.VOLUNTEER})
|
|
|
|
qs |= Channel.objects.filter(Q(tournament__in=registration.interesting_tournaments),
|
|
**{permission_type: PermissionType.TOURNAMENT_MEMBER})
|
|
|
|
qs |= Channel.objects.filter(Q(tournament__in=registration.organized_tournaments.all()),
|
|
**{permission_type: PermissionType.TOURNAMENT_ORGANIZER})
|
|
|
|
qs |= Channel.objects.filter(Q(tournament__pools__in=registration.pools_presided.all())
|
|
| Q(tournament__in=registration.organized_tournaments.all()),
|
|
**{permission_type: PermissionType.TOURNAMENT_JURY_PRESIDENT})
|
|
|
|
qs |= Channel.objects.filter(Q(pool__in=registration.jury_in.all())
|
|
| Q(pool__tournament__in=registration.organized_tournaments.all())
|
|
| Q(pool__tournament__pools__in=registration.pools_presided.all()),
|
|
**{permission_type: PermissionType.JURY_MEMBER})
|
|
|
|
qs |= Channel.objects.filter(Q(pool__in=registration.jury_in.all())
|
|
| Q(pool__tournament__in=registration.organized_tournaments.all())
|
|
| Q(pool__tournament__pools__in=registration.pools_presided.all()),
|
|
**{permission_type: PermissionType.POOL_MEMBER})
|
|
else:
|
|
registration = await ParticipantRegistration.objects \
|
|
.prefetch_related('team__participation__pools', 'team__participation__tournament').aget(user_id=user.id)
|
|
|
|
team = registration.team
|
|
tournaments = []
|
|
if team.participation.valid:
|
|
tournaments.append(team.participation.tournament)
|
|
if team.participation.final:
|
|
tournaments.append(await Tournament.objects.aget(final=True))
|
|
|
|
qs |= Channel.objects.filter(Q(tournament__in=tournaments),
|
|
**{permission_type: PermissionType.TOURNAMENT_MEMBER})
|
|
|
|
qs |= Channel.objects.filter(Q(pool__in=team.participation.pools.all()),
|
|
**{permission_type: PermissionType.POOL_MEMBER})
|
|
|
|
qs |= Channel.objects.filter(Q(team=team),
|
|
**{permission_type: PermissionType.TEAM_MEMBER})
|
|
|
|
qs |= Channel.objects.filter(invited=user).prefetch_related('invited')
|
|
|
|
return qs
|
|
|
|
class Meta:
|
|
verbose_name = _("channel")
|
|
verbose_name_plural = _("channels")
|
|
ordering = ('category', 'name',)
|
|
|
|
|
|
class Message(models.Model):
|
|
channel = models.ForeignKey(
|
|
Channel,
|
|
on_delete=models.CASCADE,
|
|
verbose_name=_("channel"),
|
|
related_name='messages',
|
|
)
|
|
|
|
author = models.ForeignKey(
|
|
'auth.User',
|
|
verbose_name=_("author"),
|
|
on_delete=models.SET_NULL,
|
|
null=True,
|
|
related_name='chat_messages',
|
|
)
|
|
|
|
created_at = models.DateTimeField(
|
|
verbose_name=_("created at"),
|
|
auto_now_add=True,
|
|
)
|
|
|
|
updated_at = models.DateTimeField(
|
|
verbose_name=_("updated at"),
|
|
auto_now=True,
|
|
)
|
|
|
|
content = models.TextField(
|
|
verbose_name=_("content"),
|
|
)
|
|
|
|
users_read = models.ManyToManyField(
|
|
'auth.User',
|
|
verbose_name=_("users read"),
|
|
related_name='+',
|
|
blank=True,
|
|
help_text=_("Users who have read the message."),
|
|
)
|
|
|
|
def get_author_name(self):
|
|
registration = self.author.registration
|
|
|
|
author_name = f"{self.author.first_name} {self.author.last_name}"
|
|
if registration.is_volunteer:
|
|
if registration.is_admin:
|
|
author_name += " (CNO)"
|
|
|
|
if self.channel.pool:
|
|
if registration == self.channel.pool.jury_president:
|
|
author_name += " (P. jury)"
|
|
elif registration in self.channel.pool.juries.all():
|
|
author_name += " (Juré⋅e)"
|
|
elif registration in self.channel.pool.tournament.organizers.all():
|
|
author_name += " (CRO)"
|
|
else:
|
|
author_name += " (Bénévole)"
|
|
elif self.channel.tournament:
|
|
if registration in self.channel.tournament.organizers.all():
|
|
author_name += " (CRO)"
|
|
elif any([registration.id == pool.jury_president
|
|
for pool in self.channel.tournament.pools.all()]):
|
|
pools = ", ".join([pool.short_name
|
|
for pool in self.channel.tournament.pools.all()
|
|
if pool.jury_president == registration])
|
|
author_name += f" (P. jury {pools})"
|
|
elif any([pool.juries.contains(registration)
|
|
for pool in self.channel.tournament.pools.all()]):
|
|
pools = ", ".join([pool.short_name
|
|
for pool in self.channel.tournament.pools.all()
|
|
if pool.juries.acontains(registration)])
|
|
author_name += f" (Juré⋅e {pools})"
|
|
else:
|
|
author_name += " (Bénévole)"
|
|
else:
|
|
if registration.organized_tournaments.exists():
|
|
tournaments = ", ".join([tournament.name
|
|
for tournament in registration.organized_tournaments.all()])
|
|
author_name += f" (CRO {tournaments})"
|
|
if Pool.objects.filter(jury_president=registration).exists():
|
|
tournaments = Tournament.objects.filter(pools__jury_president=registration).distinct()
|
|
tournaments = ", ".join([tournament.name for tournament in tournaments])
|
|
author_name += f" (P. jury {tournaments})"
|
|
elif registration.jury_in.exists():
|
|
tournaments = Tournament.objects.filter(pools__juries=registration).distinct()
|
|
tournaments = ", ".join([tournament.name for tournament in tournaments])
|
|
author_name += f" (Juré⋅e {tournaments})"
|
|
else:
|
|
if registration.team_id:
|
|
team = Team.objects.get(id=registration.team_id)
|
|
author_name += f" ({team.trigram})"
|
|
else:
|
|
author_name += " (sans équipe)"
|
|
|
|
return author_name
|
|
|
|
async def aget_author_name(self):
|
|
return await sync_to_async(self.get_author_name)()
|
|
|
|
class Meta:
|
|
verbose_name = _("message")
|
|
verbose_name_plural = _("messages")
|
|
ordering = ('created_at',)
|