Release 0.3.2 #48

Closed
Typhus wants to merge 17 commits from main into prod
29 changed files with 1179 additions and 73 deletions

View File

@ -8,13 +8,16 @@ This repository contains the code for the EZGG LAN Manager.
### Prerequisites
- Working Installation of MySQL 5 or latest MariaDB Server (`mariadb-server` for Debian-based Linux, `XAMPP` for Windows)
- Working Installation of MariaDB Server (version `10.6.25` or later)
+ MySQL should work too, but there are no guarantees.
- Python 3.9 or higher
- PyCharm or similar IDE (optional)
### Step 1: Preparing Database
To prepare the database, apply the SQL file located in `sql/create_database.sql` to your database server. This is easily accomplished with the MYSQL Workbench, but it can be also done by pipeing the file into the mariadb-server executable.
To prepare the database, apply the SQL file located in `sql/create_database.sql` to your database server. This is easily accomplished with the MYSQL Workbench, but it can be also done by piping the file into the mariadb-server executable.
After creating the database, apply all patches found in `sql/*_patch.sql` in their numeric order.
Optionally, you can now execute the script `create_demo_database_content.py`, found in `src/ezgg_lan_manager/helpers`. Be aware that it can be buggy sometimes, especially if you overwrite existing data.
@ -43,3 +46,4 @@ FLUSH PRIVILEGES;
```
3. Make sure to **NOT** use the default passwords!
4. Apply the `create_database.sql` when starting the MariaDB container for the first time.
5. Apply the patches (`sql/*_patch.sql`) when starting the MariaDB container for the first time.

View File

@ -1 +1 @@
0.2.0
0.3.2

63
sql/02-teams_patch.sql Normal file
View File

@ -0,0 +1,63 @@
-- =====================================================
-- Teams
-- =====================================================
DROP TABLE IF EXISTS `team_members`;
DROP TABLE IF EXISTS `teams`;
-- -----------------------------------------------------
-- Teams table
-- -----------------------------------------------------
CREATE TABLE `teams` (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(100) NOT NULL,
abbreviation VARCHAR(10) NOT NULL,
join_password VARCHAR(255) NOT NULL,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
UNIQUE KEY uq_team_name (name),
UNIQUE KEY uq_team_abbr (abbreviation)
) ENGINE=InnoDB
DEFAULT CHARSET=utf8mb4
COLLATE=utf8mb4_unicode_ci;
-- -----------------------------------------------------
-- Team Members (Junction Table)
-- -----------------------------------------------------
CREATE TABLE `team_members` (
team_id INT NOT NULL,
user_id INT NOT NULL,
status ENUM('MEMBER','OFFICER','LEADER')
NOT NULL DEFAULT 'MEMBER',
joined_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
PRIMARY KEY (team_id, user_id),
CONSTRAINT fk_tm_team
FOREIGN KEY (team_id)
REFERENCES teams(id)
ON DELETE CASCADE,
CONSTRAINT fk_tm_user
FOREIGN KEY (user_id)
REFERENCES users(user_id)
ON DELETE CASCADE
) ENGINE=InnoDB
DEFAULT CHARSET=utf8mb4
COLLATE=utf8mb4_unicode_ci;
-- -----------------------------------------------------
-- Indexes
-- -----------------------------------------------------
CREATE INDEX idx_tm_user
ON team_members(user_id);
CREATE INDEX idx_tm_team_status
ON team_members(team_id, status);

View File

@ -157,6 +157,12 @@ if __name__ == "__main__":
build=pages.ManageTournamentsPage,
guard=team_guard
),
ComponentPage(
name="AdminNavigationPage",
url_segment="admin",
build=pages.AdminNavigationPage,
guard=team_guard
),
ComponentPage(
name="DbErrorPage",
url_segment="db-error",
@ -171,6 +177,16 @@ if __name__ == "__main__":
name="TournamentRulesPage",
url_segment="tournament-rules",
build=pages.TournamentRulesPage,
),
ComponentPage(
name="Teams",
url_segment="teams",
build=pages.TeamsPage,
),
ComponentPage(
name="ConwaysGameOfLife",
url_segment="conway",
build=pages.ConwayPage,
)
],
theme=theme,

View File

@ -12,13 +12,14 @@ from src.ezgg_lan_manager.services.MailingService import MailingService
from src.ezgg_lan_manager.services.NewsService import NewsService
from src.ezgg_lan_manager.services.ReceiptPrintingService import ReceiptPrintingService
from src.ezgg_lan_manager.services.SeatingService import SeatingService
from src.ezgg_lan_manager.services.TeamService import TeamService
from src.ezgg_lan_manager.services.TicketingService import TicketingService
from src.ezgg_lan_manager.services.TournamentService import TournamentService
from src.ezgg_lan_manager.services.UserService import UserService
from src.ezgg_lan_manager.types import *
# Inits services in the correct order
def init_services() -> tuple[AccountingService, CateringService, ConfigurationService, DatabaseService, MailingService, NewsService, SeatingService, TicketingService, UserService, LocalDataService, ReceiptPrintingService, TournamentService]:
def init_services() -> tuple[AccountingService, CateringService, ConfigurationService, DatabaseService, MailingService, NewsService, SeatingService, TicketingService, UserService, LocalDataService, ReceiptPrintingService, TournamentService, TeamService]:
logging.basicConfig(level=logging.DEBUG)
configuration_service = ConfigurationService(from_root("config.toml"))
db_service = DatabaseService(configuration_service.get_database_configuration())
@ -32,6 +33,7 @@ def init_services() -> tuple[AccountingService, CateringService, ConfigurationSe
catering_service = CateringService(db_service, accounting_service, user_service, receipt_printing_service)
local_data_service = LocalDataService()
tournament_service = TournamentService(db_service, user_service)
team_service = TeamService(db_service)
return accounting_service, catering_service, configuration_service, db_service, mailing_service, news_service, seating_service, ticketing_service, user_service, local_data_service, receipt_printing_service, tournament_service
return accounting_service, catering_service, configuration_service, db_service, mailing_service, news_service, seating_service, ticketing_service, user_service, local_data_service, receipt_printing_service, tournament_service, team_service

Binary file not shown.

After

Width:  |  Height:  |  Size: 57 KiB

View File

@ -0,0 +1,36 @@
from rio import Component, Rectangle, Text, Column, Icon, TextStyle, PointerEventListener, PointerEvent
class AdminNavigationCard(Component):
icon_name: str
display_text: str
target_url: str
def on_press(self, _: PointerEvent) -> None:
if self.target_url:
self.session.navigate_to(self.target_url)
def build(self) -> Component:
return PointerEventListener(
Rectangle(
content=Column(
Icon(
self.icon_name,
min_width=3.5,
min_height=3.5,
fill="background",
margin=1
),
Text(self.display_text, style=TextStyle(fill=self.session.theme.background_color), justify="center", margin=1, margin_top=0)
),
cursor="pointer",
stroke_width=0.2,
stroke_color=self.session.theme.background_color,
hover_stroke_width=0.2,
hover_stroke_color=self.session.theme.hud_color,
min_width=10,
min_height=10,
corner_radius=0.2
),
on_press=self.on_press
)

View File

@ -5,6 +5,7 @@ from rio import *
from src.ezgg_lan_manager import ConfigurationService, UserService, LocalDataService
from src.ezgg_lan_manager.components.DesktopNavigationButton import DesktopNavigationButton
from src.ezgg_lan_manager.components.NavigationSponsorBox import NavigationSponsorBox
from src.ezgg_lan_manager.components.UserInfoAndLoginBox import UserInfoAndLoginBox
from src.ezgg_lan_manager.services.LocalDataService import LocalData
from src.ezgg_lan_manager.types.SessionStorage import SessionStorage
@ -42,47 +43,35 @@ class DesktopNavigation(Component):
lan_info = self.session[ConfigurationService].get_lan_info()
user_info_and_login_box = UserInfoAndLoginBox()
self.force_login_box_refresh.append(user_info_and_login_box.force_refresh)
user_navigation = [
navigation = [
DesktopNavigationButton("News", "./news"),
Spacer(min_height=1),
Spacer(min_height=0.7),
DesktopNavigationButton(f"Über {lan_info.name} {lan_info.iteration}", "./overview"),
DesktopNavigationButton("Ticket kaufen", "./buy_ticket"),
DesktopNavigationButton("Sitzplan", "./seating"),
DesktopNavigationButton("Catering", "./catering"),
DesktopNavigationButton("Teilnehmer", "./guests"),
DesktopNavigationButton("Teams", "./teams"),
DesktopNavigationButton("Turniere", "./tournaments"),
DesktopNavigationButton("FAQ", "./faq"),
DesktopNavigationButton("Regeln & AGB", "./rules-gtc"),
Spacer(min_height=1),
Spacer(min_height=0.7),
DesktopNavigationButton("Discord", "https://discord.gg/8gTjg34yyH", open_new_tab=True),
DesktopNavigationButton("Die EZ GG e.V.", "https://ezgg-ev.de/about", open_new_tab=True),
DesktopNavigationButton("Kontakt", "./contact"),
DesktopNavigationButton("Impressum & DSGVO", "./imprint"),
Spacer(min_height=1)
Spacer(min_height=0.7)
]
team_navigation = [
Text("Verwaltung", align_x=0.5, margin_top=0.3, style=TextStyle(fill=Color.from_hex("F0EADE"), font_size=1.2)),
Text("Vorsichtig sein!", align_x=0.5, margin_top=0.3, style=TextStyle(fill=self.session.theme.danger_color, font_size=0.6)),
DesktopNavigationButton("News", "./manage-news", is_team_navigation=True),
DesktopNavigationButton("Benutzer", "./manage-users", is_team_navigation=True),
DesktopNavigationButton("Catering", "./manage-catering", is_team_navigation=True),
DesktopNavigationButton("Turniere", "./manage-tournaments", is_team_navigation=True),
Spacer(min_height=1),
Revealer(
header="Normale Navigation",
content=Column(*user_navigation),
header_style=TextStyle(fill=self.session.theme.primary_color, font_size=0.9)
)
] if self.user is not None and self.user.is_team_member else []
nav_to_use = copy(team_navigation) if self.user is not None and self.user.is_team_member else copy(user_navigation)
if self.user is not None and self.user.is_team_member:
navigation.insert(0, DesktopNavigationButton("Adminbereich", "./admin", is_team_navigation=True))
return Card(
Column(
Text(lan_info.name, align_x=0.5, margin_top=0.3, style=TextStyle(fill=self.session.theme.hud_color, font_size=1.9)),
Text(f"Edition {lan_info.iteration}", align_x=0.5, style=TextStyle(fill=self.session.theme.hud_color, font_size=1.2), margin_top=0.3, margin_bottom=2),
user_info_and_login_box,
*nav_to_use,
*navigation,
Text("Unsere Sponsoren", align_x=0.5, style=TextStyle(fill=self.session.theme.hud_color, font_size=0.9), margin_bottom=0.5, margin_top=1),
NavigationSponsorBox(img_name="crackz", url="https://www.crackz.gg/"),
align_y=0
),
color=self.session.theme.neutral_color,

View File

@ -102,5 +102,5 @@ class LoginBox(Component):
min_width=12,
align_x=0.5,
margin_top=0.3,
margin_bottom=2
margin_bottom=1.5
)

View File

@ -0,0 +1,23 @@
from from_root import from_root
from rio import Component, Link, Rectangle, Image, Color
class NavigationSponsorBox(Component):
img_name: str
url: str
img_suffix: str = "png"
def build(self) -> Component:
return Link(
content=Rectangle(
content=Image(image=from_root(f"src/ezgg_lan_manager/assets/img/{self.img_name}.{self.img_suffix}"), min_width=10, min_height=10),
stroke_width=0.1,
stroke_color=Color.TRANSPARENT,
hover_stroke_width=0.1,
hover_stroke_color=self.session.theme.secondary_color,
margin=0.6,
cursor="pointer"
),
target_url=self.url,
open_in_new_tab=True
)

View File

@ -0,0 +1,44 @@
from functools import partial
from typing import Callable, Optional, Literal
from rio import Component, Revealer, TextStyle, Column, Row, Tooltip, Icon, Spacer, Text, Button
from src.ezgg_lan_manager.types.Team import TeamStatus, Team
from src.ezgg_lan_manager.types.User import User
class TeamRevealer(Component):
user: Optional[User]
team: Team
mode: Literal["join", "leave", "display"]
on_button_pressed: Callable
def build(self) -> Component:
return Revealer(
header=self.team.name,
header_style=TextStyle(
fill=self.session.theme.background_color,
font_size=1
),
content=Column(
*[Row(
Tooltip(
anchor=Icon("material/star" if self.team.members[member] == TeamStatus.LEADER else "material/stat_1", fill=self.session.theme.hud_color),
tip="Leiter" if self.team.members[member] == TeamStatus.LEADER else "Mitglied", position="top"),
Text(member.user_name, style=TextStyle(fill=self.session.theme.background_color, font_size=1), margin_left=0.5),
Spacer(grow_y=False))
for member in self.team.members
],
Row(Button(
content=f"{self.team.name} beitreten" if self.mode == "join" else f"{self.team.name} verlassen",
shape="rectangle",
style="major",
color="hud",
on_press=partial(self.on_button_pressed, self.team),
), margin_top=1, margin_bottom=1),
margin_right=1,
margin_left=1
),
margin_left=1,
margin_right=1,
)

View File

@ -0,0 +1,222 @@
import logging
from typing import Optional, Callable
from rio import Component, Text, Spacer, Rectangle, Column, TextStyle, Row, Button, TextInput, ThemeContextSwitcher
from src.ezgg_lan_manager.services.TeamService import TeamService, NotMemberError, TeamLeadRemovalError, AlreadyMemberError, NameNotAllowedError, TeamNameTooLongError, \
TeamAbbrInvalidError, TeamNameAlreadyTaken
from src.ezgg_lan_manager.types.Team import Team
from src.ezgg_lan_manager.types.User import User
logger = logging.getLogger(__name__.split(".")[-1])
class ErrorBox(Component):
error_message: str
cancel: Callable
def build(self) -> Component:
return Rectangle(
content=Column(
Text(self.error_message, style=TextStyle(fill=self.session.theme.background_color), margin_bottom=1.5),
Row(
Button(
content="Ok",
shape="rectangle",
style="major",
color="hud",
on_press=self.cancel,
)
),
margin=1
),
fill=self.session.theme.primary_color
)
class TeamsDialogJoinHandler(Component):
is_active: bool
cancel: Callable
user: Optional[User] = None
team: Optional[Team] = None
error_message: Optional[str] = None
password: str = ""
async def join(self) -> None:
if self.user is None or self.team is None:
return
if self.password != self.team.join_password:
self.error_message = "Falsches Passwort!"
return
try:
await self.session[TeamService].add_member_to_team(self.team, self.user)
except AlreadyMemberError:
self.error_message = "Du bist bereits Mitglied dieses Teams"
else:
await self.cancel_with_reset()
async def cancel_with_reset(self) -> None:
await self.cancel()
self.error_message = None
self.password = ""
def build(self) -> Component:
if not self.is_active or self.user is None or self.team is None:
return Spacer()
if self.error_message is not None:
return ErrorBox(error_message=self.error_message, cancel=self.cancel_with_reset)
return Rectangle(
content=Column(
Text(f"Team {self.team.name} beitreten", style=TextStyle(fill=self.session.theme.background_color), margin_bottom=1, justify="center"),
ThemeContextSwitcher(content=TextInput(text=self.bind().password, label="Beitrittspasswort", margin_bottom=1), color="secondary"),
Row(
Button(
content="Abbrechen",
shape="rectangle",
style="major",
color=self.session.theme.danger_color,
on_press=self.cancel_with_reset,
),
Button(
content="Beitreten",
shape="rectangle",
style="major",
color=self.session.theme.success_color,
on_press=self.join,
),
spacing=1
),
margin=1
),
fill=self.session.theme.primary_color
)
class TeamsDialogLeaveHandler(Component):
is_active: bool
cancel: Callable
user: Optional[User] = None
team: Optional[Team] = None
error_message: Optional[str] = None
async def leave(self) -> None:
if self.user is not None and self.team is not None:
try:
await self.session[TeamService].remove_member_from_team(self.team, self.user)
except NotMemberError:
self.error_message = "Du bist kein Mitglied in diesem Team"
except TeamLeadRemovalError:
self.error_message = "Als Teamleiter kannst du das Team nicht verlassen"
else:
await self.cancel_with_reset()
async def cancel_with_reset(self) -> None:
await self.cancel()
self.error_message = None
def build(self) -> Component:
if not self.is_active or self.user is None or self.team is None:
return Spacer()
if self.error_message is not None:
return ErrorBox(error_message=self.error_message, cancel=self.cancel_with_reset)
return Rectangle(
content=Column(
Text(f"Team {self.team.name} wirklich verlassen?", style=TextStyle(fill=self.session.theme.background_color), margin_bottom=1.5, justify="center"),
Row(
Button(
content="Nein",
shape="rectangle",
style="major",
color=self.session.theme.danger_color,
on_press=self.cancel_with_reset,
),
Button(
content="Ja",
shape="rectangle",
style="major",
color=self.session.theme.success_color,
on_press=self.leave,
),
spacing=1
),
margin=1
),
fill=self.session.theme.primary_color
)
class TeamsDialogCreateHandler(Component):
is_active: bool
cancel: Callable
user: Optional[User] = None
error_message: Optional[str] = None
team_name: str = ""
team_abbr: str = ""
team_join_password: str = ""
async def cancel_with_reset(self) -> None:
await self.cancel()
self.error_message = None
self.team_name, self.team_abbr, self.team_join_password = "", "", ""
async def create(self) -> None:
if self.user is None:
return
if not self.team_name or not self.team_abbr or not self.team_join_password:
self.error_message = "Angaben unvollständig"
return
try:
await self.session[TeamService].create_team(self.team_name, self.team_abbr, self.team_join_password, self.user)
except NameNotAllowedError as e:
self.error_message = f"Angaben ungültig. Darf kein '{e.disallowed_char}' enthalten."
except TeamNameTooLongError:
self.error_message = f"Name zu lang. Maximal {TeamService.MAX_TEAM_NAME_LENGTH} Zeichen."
except TeamAbbrInvalidError:
self.error_message = f"Name zu lang. Maximal {TeamService.MAX_TEAM_ABBR_LENGTH} Zeichen."
except TeamNameAlreadyTaken:
self.error_message = "Ein Team mit diesem Namen existiert bereits."
else:
await self.cancel_with_reset()
def build(self) -> Component:
if not self.is_active or self.user is None:
return Spacer()
if self.error_message is not None:
return ErrorBox(error_message=self.error_message, cancel=self.cancel_with_reset)
return Rectangle(
content=Column(
Text(f"Team gründen", style=TextStyle(fill=self.session.theme.background_color), margin_bottom=1.5, justify="center"),
ThemeContextSwitcher(content=TextInput(text=self.bind().team_name, label="Team Name", margin_bottom=1), color="secondary"),
ThemeContextSwitcher(content=TextInput(text=self.bind().team_abbr, label="Team Abkürzung", margin_bottom=1), color="secondary"),
ThemeContextSwitcher(content=TextInput(text=self.bind().team_join_password, label="Beitrittspasswort", margin_bottom=1), color="secondary"),
Row(
Button(
content="Abbrechen",
shape="rectangle",
style="major",
color=self.session.theme.danger_color,
on_press=self.cancel_with_reset,
),
Button(
content="Gründen",
shape="rectangle",
style="major",
color=self.session.theme.success_color,
on_press=self.create,
),
spacing=1
),
margin=1
),
fill=self.session.theme.primary_color
)

View File

@ -117,5 +117,5 @@ class UserInfoBox(Component):
min_width=12,
align_x=0.5,
margin_top=0.3,
margin_bottom=2
margin_bottom=1.5
)

View File

@ -0,0 +1,44 @@
from rio import Column, Component, event, Text, TextStyle, Row
from src.ezgg_lan_manager.components.AdminNavigationCard import AdminNavigationCard
from src.ezgg_lan_manager.components.MainViewContentBox import MainViewContentBox
from src.ezgg_lan_manager.services.ConfigurationService import ConfigurationService
class AdminNavigationPage(Component):
@event.on_populate
async def on_populate(self) -> None:
await self.session.set_title(f"{self.session[ConfigurationService].get_lan_info().name} - Admin")
def build(self) -> Component:
return Column(
MainViewContentBox(
Text(
text="Admin",
style=TextStyle(
fill=self.session.theme.background_color,
font_size=1.2
),
margin_top=2,
margin_bottom=2,
align_x=0.5
)
),
MainViewContentBox(
Column(
Row(
AdminNavigationCard(icon_name="material/supervised_user_circle", display_text="Nutzer", target_url="manage-users"),
AdminNavigationCard(icon_name="material/fastfood", display_text="Catering", target_url="manage-catering"),
spacing=1
),
Row(
AdminNavigationCard(icon_name="material/text_ad", display_text="News", target_url="manage-news"),
AdminNavigationCard(icon_name="material/trophy", display_text="Turniere", target_url="manage-tournaments"),
spacing=1
),
margin=1,
spacing=1
)
),
align_y=0
)

View File

@ -2,7 +2,7 @@ from __future__ import annotations
from typing import * # type: ignore
from rio import Component, event, Spacer, Card, Container, Column, Row, TextStyle, Color, Text, PageView, Button
from rio import Component, event, Spacer, Card, Container, Column, Row, TextStyle, Color, Text, PageView, Button, Link
from src.ezgg_lan_manager import ConfigurationService, DatabaseService
from src.ezgg_lan_manager.components.DesktopNavigation import DesktopNavigation
@ -58,7 +58,15 @@ class BasePage(Component):
Row(
Spacer(grow_x=True, grow_y=False),
Card(
content=Text(f"EZGG LAN Manager Version {self.session[ConfigurationService].APP_VERSION} © EZ GG e.V.", align_x=0.5, align_y=0.5, fill=self.session.theme.primary_color, style=TextStyle(font_size=0.5)),
content=Row(
Text(f"EZGG LAN Manager Version {self.session[ConfigurationService].APP_VERSION} © EZ GG e.V.", fill=self.session.theme.primary_color, style=TextStyle(font_size=0.5)),
Text(f"-", fill=self.session.theme.primary_color, style=TextStyle(font_size=0.5), margin_left=0.5, margin_right=0.5),
Link(content=Text(f"Impressum & DSGVO", fill=self.session.theme.primary_color, style=TextStyle(font_size=0.5)), target_url="./imprint"),
Text(f"-", fill=self.session.theme.primary_color, style=TextStyle(font_size=0.5), margin_left=0.5, margin_right=0.5),
Link(content=Text(f"Kontakt", fill=self.session.theme.primary_color, style=TextStyle(font_size=0.5)), target_url="./contact"),
align_x=0.5,
align_y=0.5
),
color=self.session.theme.neutral_color,
corner_radius=(0, 0, 0.5, 0.5),
grow_x=False,

View File

@ -19,14 +19,18 @@ class BuyTicketPage(Component):
popup_message: str = ""
is_popup_success: bool = False
is_buying_enabled: bool = False
is_user_logged_in: bool = False
@event.on_populate
async def on_populate(self) -> None:
self.session[SessionStorage].subscribe_to_logged_in_or_out_event(str(self.__class__), self.on_populate)
await self.session.set_title(f"{self.session[ConfigurationService].get_lan_info().name} - Ticket kaufen")
self.user = await self.session[UserService].get_user(self.session[SessionStorage].user_id)
if self.user is None: # No user logged in
self.is_buying_enabled = False
self.is_user_logged_in = False
else: # User is logged in
self.is_user_logged_in = True
possible_ticket = await self.session[TicketingService].get_user_ticket(self.user.user_id)
self.user_ticket = possible_ticket
if possible_ticket is not None: # User already has a ticket
@ -67,17 +71,29 @@ class BuyTicketPage(Component):
def build(self) -> Component:
ticket_infos = self.session[ConfigurationService].get_ticket_info()
header = Text(
"Tickets & Preise",
style=TextStyle(
fill=self.session.theme.background_color,
font_size=1.2
header = Column(
Text(
"Tickets & Preise",
style=TextStyle(
fill=self.session.theme.background_color,
font_size=1.2
),
margin_top=2,
align_x=0.5
),
margin_top=2,
margin_bottom=2,
align_x=0.5
spacing=0.2
)
if not self.is_user_logged_in:
header.add(Text(
"Du musst eingeloggt sein\num ein Ticket zu kaufen",
style=TextStyle(
fill=self.session.theme.background_color,
font_size=0.6
),
align_x=0.5
))
return Column(
MainViewContentBox(
Column(

View File

@ -0,0 +1,83 @@
from __future__ import annotations
from copy import deepcopy
from random import randint
from typing import * # type: ignore
from rio import Component, event, Column, Row, Color, Rectangle
class ConwayPage(Component):
"""
This is an Easter egg.
"""
active_generation: list[list] = []
rows: int = 36
cols: int = 20
@event.periodic(1)
async def calc_next_gen(self) -> None:
self.create_next_grid()
@event.on_populate
def prepare(self) -> None:
self.active_generation = self.create_initial_grid()
def create_initial_grid(self) -> list[list]:
grid = []
for row in range(self.rows):
grid_rows = []
for col in range(self.cols):
if randint(0, 7) == 0:
grid_rows += [1]
else:
grid_rows += [0]
grid += [grid_rows]
return grid
def create_next_grid(self) -> None:
next_grid = deepcopy(self.active_generation)
for row in range(self.rows):
for col in range(self.cols):
live_neighbors = self.get_live_neighbors(row, col, self.active_generation)
if live_neighbors < 2 or live_neighbors > 3:
next_grid[row][col] = 0
elif live_neighbors == 3 and self.active_generation[row][col] == 0:
next_grid[row][col] = 1
else:
next_grid[row][col] = self.active_generation[row][col]
self.active_generation = next_grid
def get_live_neighbors(self, row: int, col: int, grid: list[list]) -> int:
life_sum = 0
for i in range(-1, 2):
for j in range(-1, 2):
if not (i == 0 and j == 0):
life_sum += grid[((row + i) % self.rows)][((col + j) % self.cols)]
return life_sum
def grid_changing(self, next_grid: list[list]) -> bool:
for row in range(self.rows):
for col in range(self.cols):
if not self.active_generation[row][col] == next_grid[row][col]:
return True
return False
def build(self) -> Component:
rows = []
for row in self.active_generation:
rectangles = []
for cell in row:
color = Color.WHITE if cell == 1 else Color.BLACK
rectangles.append(Rectangle(fill=color, transition_time=0.3))
rows.append(Row(*rectangles))
return Column(*rows)

View File

@ -1,5 +1,6 @@
import logging
from datetime import datetime
from functools import partial
from typing import Optional
from from_root import from_root
@ -40,7 +41,11 @@ class ManageTournamentsPage(Component):
return
users = await self.session[UserService].get_all_users()
try:
self.cancel_options = {next(filter(lambda u: u.user_id == p.id, users)).user_name: p for p in tournament.participants}
cancel_options = {next(filter(lambda u: u.user_id == p.id, users)).user_name: p for p in tournament.participants}
if cancel_options:
self.cancel_options = cancel_options
else:
self.cancel_options = {"": None}
except StopIteration as e:
logger.error(f"Error trying to find user for participant: {e}")
self.tournament_id_selected_for_participant_removal = tournament_id
@ -70,13 +75,12 @@ class ManageTournamentsPage(Component):
Text(tournament.name, style=TextStyle(fill=self.session.theme.background_color, font_size=0.8), justify="left", margin_right=1.5),
Text(f"{weekday_to_display_text(tournament.start_time.weekday())[:2]}.{tournament.start_time.strftime('%H:%M')} Uhr", style=TextStyle(fill=start_time_color, font_size=0.8), justify="left", margin_right=1),
Spacer(),
Tooltip(anchor=IconButton("material/play_arrow", min_size=2, margin_right=0.5, on_press=lambda: self.on_start_pressed(tournament.id)), tip="Starten"),
Tooltip(anchor=IconButton("material/cancel_schedule_send", min_size=2, margin_right=0.5, on_press=lambda: self.on_cancel_pressed(tournament.id)), tip="Absagen"),
Tooltip(anchor=IconButton("material/person_cancel", min_size=2, on_press=lambda: self.on_remove_participant_pressed(tournament.id)), tip="Spieler entfernen"),
Tooltip(anchor=IconButton("material/play_arrow", min_size=2, margin_right=0.5, on_press=partial(self.on_start_pressed, tournament.id)), tip="Starten"),
Tooltip(anchor=IconButton("material/cancel_schedule_send", min_size=2, margin_right=0.5, on_press=partial(self.on_cancel_pressed, tournament.id)), tip="Absagen"),
Tooltip(anchor=IconButton("material/person_cancel", min_size=2, on_press=partial(self.on_remove_participant_pressed, tournament.id)), tip="Spieler entfernen"),
margin=1
)
)
return Column(
Popup(
anchor=MainViewContentBox(

View File

@ -3,7 +3,7 @@ from dataclasses import field
from typing import Optional
from rio import Column, Component, event, TextStyle, Text, TextInput, ThemeContextSwitcher, Grid, \
PointerEventListener, PointerEvent, Rectangle, CursorStyle, Color, TextInputChangeEvent, Spacer, Row, Switch, \
PointerEventListener, PointerEvent, Rectangle, Color, TextInputChangeEvent, Spacer, Row, Switch, \
SwitchChangeEvent, EventHandler
from src.ezgg_lan_manager import ConfigurationService, UserService, AccountingService, SeatingService, MailingService
@ -42,7 +42,7 @@ class ClickableGridContent(Component):
grow_x=True
),
fill=Color.TRANSPARENT,
cursor=CursorStyle.POINTER
cursor="pointer"
),
on_pointer_enter=self.on_mouse_enter,
on_pointer_leave=self.on_mouse_leave,

View File

@ -75,7 +75,7 @@ class OverviewPage(Component):
Row(
Text("Internet", fill=self.session.theme.neutral_color, margin_left=1),
Spacer(),
Text(f"60/20 Mbit/s (down/up)", fill=self.session.theme.neutral_color, margin_right=1),
Text(f"100/50 Mbit/s (down/up)", fill=self.session.theme.neutral_color, margin_right=1),
margin_bottom=0.3
),
Row(
@ -125,17 +125,5 @@ class OverviewPage(Component):
)
)
),
MainViewContentBox(
Column(
Text("Turniere & Ablauf", font_size=2, justify="center", fill=self.session.theme.neutral_color, margin_top=0.5, margin_bottom=1),
Column(
Row(
Text("Zum aktuellen Zeitpunkt steht noch nicht fest welche Turniere gespielt werden. Wir planen diverse Online- und Offline Turniere mit Preisen durchzuführen. Weitere Informationen gibt es, sobald sie kommen, auf der NEWS- und Turnier-Seite.", font_size=0.7,
fill=self.session.theme.neutral_color, margin_left=1, overflow="wrap"),
margin_bottom=0.3
)
)
)
),
Spacer()
)

View File

@ -0,0 +1,172 @@
from asyncio import sleep
from rio import event, ProgressCircle, PointerEventListener, PointerEvent, Popup, Color
from src.ezgg_lan_manager import ConfigurationService
from src.ezgg_lan_manager.components.MainViewContentBox import MainViewContentBox
from src.ezgg_lan_manager.components.TeamRevealer import TeamRevealer
from src.ezgg_lan_manager.components.TeamsDialogHandler import *
from src.ezgg_lan_manager.services.TeamService import TeamService
from src.ezgg_lan_manager.services.UserService import UserService
from src.ezgg_lan_manager.types.SessionStorage import SessionStorage
from src.ezgg_lan_manager.types.Team import Team
from src.ezgg_lan_manager.types.User import User
class TeamsPage(Component):
all_teams: Optional[list[Team]] = None
user: Optional[User] = None
# Dialog handling
popup_open: bool = False
join_active: bool = False
leave_active: bool = True
create_active: bool = False
selected_team_for_join_or_leave: Optional[Team] = None
@event.on_populate
async def on_populate(self) -> None:
self.all_teams = await self.session[TeamService].get_all_teams()
self.user = await self.session[UserService].get_user(self.session[SessionStorage].user_id)
await self.session.set_title(f"{self.session[ConfigurationService].get_lan_info().name} - Teams")
self.session[SessionStorage].subscribe_to_logged_in_or_out_event(str(self.__class__), self.on_populate)
async def on_join_button_pressed(self, team: Team) -> None:
if self.user is None:
return
self.selected_team_for_join_or_leave = team
self.join_active, self.leave_active, self.create_active = True, False, False
self.popup_open = True
async def on_leave_button_pressed(self, team: Team) -> None:
if self.user is None:
return
self.selected_team_for_join_or_leave = team
self.join_active, self.leave_active, self.create_active = False, True, False
self.popup_open = True
async def on_create_button_pressed(self, _: PointerEvent) -> None:
if self.user is None:
return
self.join_active, self.leave_active, self.create_active = False, False, True
self.popup_open = True
async def popup_action_cancelled(self) -> None:
self.popup_open = False
await sleep(0.2) # Waits for the animation to play before resetting its contents
self.join_active, self.leave_active, self.create_active = False, False, False
self.selected_team_for_join_or_leave = None
self.all_teams = await self.session[TeamService].get_all_teams()
def build(self) -> Component:
if self.all_teams is None:
return Column(
MainViewContentBox(
ProgressCircle(
color="secondary",
align_x=0.5,
margin_top=1,
margin_bottom=1
)
),
Spacer()
)
team_list = []
for team in self.all_teams:
team_list.append(
TeamRevealer(
user=self.user,
team=team,
mode="leave" if self.user in team.members.keys() else "join",
on_button_pressed=self.on_leave_button_pressed if self.user in team.members.keys() else self.on_join_button_pressed
)
)
if team_list:
team_list[-1].margin_bottom = 1
own_teams_content = Spacer(grow_x=False, grow_y=False)
if self.user is not None:
user_team_list = []
for team in self.all_teams:
if self.user in team.members.keys():
user_team_list.append(TeamRevealer(user=self.user, team=team, mode="leave", on_button_pressed=self.on_leave_button_pressed))
if not user_team_list:
user_team_list.append(Text(
text="Du bist noch in keinem Team.",
style=TextStyle(
fill=self.session.theme.background_color,
font_size=1
),
margin_top=1,
margin_bottom=1,
align_x=0.5
))
else:
user_team_list[-1].margin_bottom = 1
own_teams_content = MainViewContentBox(
Column(
Row(
Text(
text="Deine Teams",
style=TextStyle(
fill=self.session.theme.background_color,
font_size=1.2
),
grow_x=True,
justify="right",
margin_right=3
),
Column(
PointerEventListener(Rectangle(
content=Text(text="Team erstellen", style=TextStyle(fill=self.session.theme.background_color, font_size=0.7), margin=0.1, selectable=False),
stroke_width=0.1,
stroke_color=self.session.theme.hud_color,
cursor="pointer",
hover_fill=self.session.theme.hud_color,
transition_time=0
), on_press=self.on_create_button_pressed),
Spacer(),
margin_right=2
),
margin_top=1,
margin_bottom=1
),
*user_team_list
)
)
return Popup(
anchor=Column(
own_teams_content,
MainViewContentBox(
Column(
Text(
text="Alle Teams",
style=TextStyle(
fill=self.session.theme.background_color,
font_size=1.2
),
margin_top=1,
margin_bottom=1,
align_x=0.5
),
*team_list
)
),
align_y=0,
),
content=Column(
TeamsDialogJoinHandler(is_active=self.join_active, cancel=self.popup_action_cancelled, user=self.user, team=self.selected_team_for_join_or_leave),
TeamsDialogLeaveHandler(is_active=self.leave_active, cancel=self.popup_action_cancelled, user=self.user, team=self.selected_team_for_join_or_leave),
TeamsDialogCreateHandler(is_active=self.create_active, cancel=self.popup_action_cancelled, user=self.user)
),
is_open=self.popup_open,
modal=False,
corner_radius=(0.5, 0.5, 0.5, 0.5),
color=Color.TRANSPARENT,
user_closable=False,
position="top"
)

View File

@ -1,14 +1,14 @@
from asyncio import sleep
from typing import Optional, Union, Literal
from from_root import from_root
from rio import Column, Component, event, TextStyle, Text, Row, Image, Spacer, ProgressCircle, Button, Checkbox, ThemeContextSwitcher, Link, Revealer, PointerEventListener, \
PointerEvent, Rectangle, Color
from src.ezgg_lan_manager import ConfigurationService, TournamentService, UserService
from src.ezgg_lan_manager import ConfigurationService, TournamentService, UserService, TicketingService
from src.ezgg_lan_manager.components.MainViewContentBox import MainViewContentBox
from src.ezgg_lan_manager.components.TournamentDetailsInfoRow import TournamentDetailsInfoRow
from src.ezgg_lan_manager.types.DateUtil import weekday_to_display_text
from src.ezgg_lan_manager.types.Participant import Participant
from src.ezgg_lan_manager.types.SessionStorage import SessionStorage
from src.ezgg_lan_manager.types.Tournament import Tournament
from src.ezgg_lan_manager.types.TournamentBase import TournamentStatus, tournament_status_to_display_text, tournament_format_to_display_texts
@ -45,6 +45,14 @@ class TournamentDetailsPage(Component):
self.loading_done()
@staticmethod
async def artificial_delay() -> None:
await sleep(0.8) # https://medium.com/design-bootcamp/ux-psychology-of-artificial-waiting-enhancing-user-experiences-through-deliberate-delays-d7822faf3930
async def update(self) -> None:
self.tournament = await self.session[TournamentService].get_tournament_by_id(self.tournament.id)
self.current_tournament_user_list = await self.session[TournamentService].get_users_from_participant_list(self.tournament.participants)
def open_close_participant_revealer(self, _: PointerEvent) -> None:
self.participant_revealer_open = not self.participant_revealer_open
@ -53,15 +61,21 @@ class TournamentDetailsPage(Component):
if not self.user:
return
try:
await self.session[TournamentService].register_user_for_tournament(self.user.user_id, self.tournament.id)
self.is_success = True
self.message = f"Erfolgreich angemeldet!"
except Exception as e:
user_ticket = await self.session[TicketingService].get_user_ticket(self.user.user_id)
if user_ticket is None:
self.is_success = False
self.message = f"Fehler: {e}"
self.message = "Turnieranmeldung nur mit Ticket"
else:
try:
await self.session[TournamentService].register_user_for_tournament(self.user.user_id, self.tournament.id)
await self.artificial_delay()
self.is_success = True
self.message = f"Erfolgreich angemeldet!"
except Exception as e:
self.is_success = False
self.message = f"Fehler: {e}"
await self.update()
self.loading = False
await self.on_populate()
async def unregister_pressed(self) -> None:
self.loading = True
@ -70,13 +84,14 @@ class TournamentDetailsPage(Component):
try:
await self.session[TournamentService].unregister_user_from_tournament(self.user.user_id, self.tournament.id)
await self.artificial_delay()
self.is_success = True
self.message = f"Erfolgreich abgemeldet!"
except Exception as e:
self.is_success = False
self.message = f"Fehler: {e}"
await self.update()
self.loading = False
await self.on_populate()
async def tree_button_clicked(self) -> None:
pass # ToDo: Implement tournament tree view
@ -198,7 +213,7 @@ class TournamentDetailsPage(Component):
content=Rectangle(
content=TournamentDetailsInfoRow(
"Teilnehmer ▴" if self.participant_revealer_open else "Teilnehmer ▾",
f"{len(self.tournament.participants)} / {self.tournament.max_participants}",
f"{len(self.current_tournament_user_list)} / {self.tournament.max_participants}",
value_color=self.session.theme.danger_color if self.tournament.is_full else self.session.theme.background_color,
key_color=self.session.theme.secondary_color
),

View File

@ -22,3 +22,6 @@ from .ManageTournamentsPage import ManageTournamentsPage
from .OverviewPage import OverviewPage
from .TournamentDetailsPage import TournamentDetailsPage
from .TournamentRulesPage import TournamentRulesPage
from .ConwayPage import ConwayPage
from .TeamsPage import TeamsPage
from .AdminNavigationPage import AdminNavigationPage

View File

@ -14,6 +14,7 @@ from src.ezgg_lan_manager.types.ConfigurationTypes import DatabaseConfiguration
from src.ezgg_lan_manager.types.News import News
from src.ezgg_lan_manager.types.Participant import Participant
from src.ezgg_lan_manager.types.Seat import Seat
from src.ezgg_lan_manager.types.Team import TeamStatus, Team
from src.ezgg_lan_manager.types.Ticket import Ticket
from src.ezgg_lan_manager.types.Tournament import Tournament
from src.ezgg_lan_manager.types.TournamentBase import GameTitle, TournamentFormat, TournamentStatus, ParticipantType
@ -967,3 +968,208 @@ class DatabaseService:
return await self.remove_participant_from_tournament(participant, tournament)
except Exception as e:
logger.warning(f"Error removing participant from tournament: {e}")
async def get_teams(self) -> list[Team]:
async with self._connection_pool.acquire() as conn:
async with conn.cursor(aiomysql.Cursor) as cursor:
query = """
SELECT
t.id AS team_id,
t.name AS team_name,
t.abbreviation AS team_abbr,
t.join_password,
t.created_at AS team_created_at,
tm.status AS team_status,
tm.joined_at AS member_joined_at,
u.*
FROM teams t
LEFT JOIN team_members tm
ON t.id = tm.team_id
LEFT JOIN users u
ON tm.user_id = u.user_id
ORDER BY
t.id,
CASE tm.status
WHEN 'LEADER' THEN 1
WHEN 'OFFICER' THEN 2
WHEN 'MEMBER' THEN 3
ELSE 4
END,
u.user_name;
"""
try:
await cursor.execute(query)
await conn.commit()
except aiomysql.InterfaceError:
pool_init_result = await self.init_db_pool()
if not pool_init_result:
raise NoDatabaseConnectionError
return await self.get_teams()
except Exception as e:
logger.warning(f"Error getting teams: {e}")
return []
current_team: Optional[Team] = None
all_teams = []
for row in await cursor.fetchall():
if row[5] is None: # Teams without single member are ignored
continue
if current_team is None:
user = self._map_db_result_to_user(row[7:])
current_team = Team(id=row[0], name=row[1], abbreviation=row[2], join_password=row[3], members={user: TeamStatus.from_str(row[5])})
elif current_team.id == row[0]: # Still same team
current_team.members[self._map_db_result_to_user(row[7:])] = TeamStatus.from_str(row[5])
else:
all_teams.append(current_team)
user = self._map_db_result_to_user(row[7:])
current_team = Team(id=row[0], name=row[1], abbreviation=row[2], join_password=row[3], members={user: TeamStatus.from_str(row[5])})
all_teams.append(current_team)
return all_teams
async def get_team_by_id(self, team_id: int) -> Optional[Team]:
async with self._connection_pool.acquire() as conn:
async with conn.cursor(aiomysql.Cursor) as cursor:
query = """
SELECT
t.id AS team_id,
t.name AS team_name,
t.abbreviation AS team_abbr,
t.join_password,
t.created_at AS team_created_at,
tm.status AS team_status,
tm.joined_at AS member_joined_at,
u.*
FROM teams t
LEFT JOIN team_members tm
ON t.id = tm.team_id
LEFT JOIN users u
ON tm.user_id = u.user_id
WHERE t.id = %s
ORDER BY
t.id,
CASE tm.status
WHEN 'LEADER' THEN 1
WHEN 'OFFICER' THEN 2
WHEN 'MEMBER' THEN 3
ELSE 4
END,
u.user_name;
"""
try:
await cursor.execute(query, (team_id, ))
await conn.commit()
except aiomysql.InterfaceError:
pool_init_result = await self.init_db_pool()
if not pool_init_result:
raise NoDatabaseConnectionError
return await self.get_team_by_id(team_id)
except Exception as e:
logger.warning(f"Error getting team: {e}")
return None
team: Optional[Team] = None
for row in await cursor.fetchall():
if team is None:
user = self._map_db_result_to_user(row[7:])
team = Team(id=row[0], name=row[1], abbreviation=row[2], join_password=row[3], members={user: TeamStatus.from_str(row[5])})
elif team.id == row[0]:
team.members[self._map_db_result_to_user(row[7:])] = TeamStatus.from_str(row[5])
return team
async def create_team(self, team_name: str, team_abbr: str, join_password: str, leader: User) -> Team:
async with self._connection_pool.acquire() as conn:
async with conn.cursor(aiomysql.Cursor) as cursor:
try:
await cursor.execute(
"INSERT INTO teams (name, abbreviation, join_password) "
"VALUES (%s, %s, %s)", (team_name, team_abbr, join_password)
)
await conn.commit()
team_id = cursor.lastrowid
await cursor.execute(
"INSERT INTO team_members (team_id, user_id, status) VALUES (%s, %s, %s)",
(team_id, leader.user_id, TeamStatus.LEADER.name)
)
await conn.commit()
return await self.get_team_by_id(team_id)
except aiomysql.InterfaceError:
pool_init_result = await self.init_db_pool()
if not pool_init_result:
raise NoDatabaseConnectionError
return await self.create_team(team_name, team_abbr, join_password)
except aiomysql.IntegrityError as e:
logger.warning(f"Aborted duplication entry: {e}")
raise DuplicationError
async def update_team(self, team: Team) -> Team:
async with self._connection_pool.acquire() as conn:
async with conn.cursor(aiomysql.Cursor) as cursor:
try:
await cursor.execute(
"UPDATE teams SET name = %s, abbreviation = %s, join_password = %s WHERE (id = %s)",
(team.name, team.abbreviation, team.join_password, team.id)
)
await conn.commit()
return await self.get_team_by_id(team.id)
except aiomysql.InterfaceError:
pool_init_result = await self.init_db_pool()
if not pool_init_result:
raise NoDatabaseConnectionError
return await self.update_team(team)
except aiomysql.IntegrityError as e:
logger.warning(f"Aborted duplication entry: {e}")
raise DuplicationError
async def add_member_to_team(self, team: Team, user: User, status: TeamStatus = TeamStatus.MEMBER) -> None:
async with self._connection_pool.acquire() as conn:
async with conn.cursor(aiomysql.Cursor) as cursor:
try:
await cursor.execute(
"INSERT INTO team_members (team_id, user_id, status) VALUES (%s, %s, %s)",
(team.id, user.user_id, status.name)
)
await conn.commit()
except aiomysql.InterfaceError:
pool_init_result = await self.init_db_pool()
if not pool_init_result:
raise NoDatabaseConnectionError
return await self.add_member_to_team(team, user, status)
except aiomysql.IntegrityError as e:
logger.warning(f"Failed to add member {user.user_name} to team {team.name}: {e}")
raise DuplicationError
async def remove_user_from_team(self, team: Team, user: User) -> None:
async with self._connection_pool.acquire() as conn:
async with conn.cursor(aiomysql.Cursor) as cursor:
try:
await cursor.execute(
"DELETE FROM team_members WHERE team_id = %s AND user_id = %s",
(team.id, user.user_id)
)
await conn.commit()
except aiomysql.InterfaceError:
pool_init_result = await self.init_db_pool()
if not pool_init_result:
raise NoDatabaseConnectionError
return await self.remove_user_from_team(team, user)

View File

@ -45,7 +45,7 @@ class MailingService:
return f"""
Hallo {user.user_name},
deinem Account wurden {added_balance} hinzugefügt. Dein neues Guthaben beträgt nun {total_balance} .
deinem Account wurden {added_balance:.2f} hinzugefügt. Dein neues Guthaben beträgt nun {total_balance:.2f} .
Wenn du zu dieser Aufladung Fragen hast, stehen wir dir in unserem Discord Server oder per Mail an {self._configuration_service.get_lan_info().organizer_mail} zur Verfügung.

View File

@ -0,0 +1,134 @@
from string import ascii_letters, digits
from typing import Optional
from src.ezgg_lan_manager.services.DatabaseService import DatabaseService, DuplicationError
from src.ezgg_lan_manager.types.Team import TeamStatus, Team
from src.ezgg_lan_manager.types.User import User
class NameNotAllowedError(Exception):
def __init__(self, disallowed_char: str) -> None:
self.disallowed_char = disallowed_char
class AlreadyMemberError(Exception):
def __init__(self) -> None:
pass
class NotMemberError(Exception):
def __init__(self) -> None:
pass
class TeamLeadRemovalError(Exception):
def __init__(self) -> None:
pass
class TeamNameTooLongError(Exception):
def __init__(self) -> None:
pass
class TeamNameAlreadyTaken(Exception):
def __init__(self) -> None:
pass
class TeamAbbrInvalidError(Exception):
def __init__(self) -> None:
pass
class TeamService:
ALLOWED_TEAM_NAME_SYMBOLS = ascii_letters + digits + "!#$%&*+,-./:;<=>?[]^_{|}~ "
MAX_TEAM_NAME_LENGTH = 24
MAX_TEAM_ABBR_LENGTH = 8
def __init__(self, db_service: DatabaseService) -> None:
self._db_service = db_service
async def get_all_teams(self) -> list[Team]:
return await self._db_service.get_teams()
async def get_team_by_id(self, team_id: int) -> Optional[Team]:
return await self._db_service.get_team_by_id(team_id)
async def get_teams_for_user_by_id(self, user_id: int) -> list[Team]:
all_teams = await self.get_all_teams()
user_teams = []
for team in all_teams:
if user_id in [u.user_id for u in team.members.keys()]:
user_teams.append(team)
return user_teams
async def create_team(self, team_name: str, team_abbr: str, join_password: str, leader: User) -> Team:
disallowed_char = self._check_for_disallowed_char(team_name)
if disallowed_char:
raise NameNotAllowedError(disallowed_char)
disallowed_char = self._check_for_disallowed_char(team_abbr)
if disallowed_char:
raise NameNotAllowedError(disallowed_char)
if not team_name or len(team_name) > self.MAX_TEAM_NAME_LENGTH:
raise TeamNameTooLongError()
if not team_abbr or len(team_abbr) > self.MAX_TEAM_ABBR_LENGTH:
raise TeamAbbrInvalidError()
try:
created_team = await self._db_service.create_team(team_name, team_abbr, join_password, leader)
except DuplicationError:
raise TeamNameAlreadyTaken
return created_team
async def update_team(self, team: Team) -> Team:
"""
Updates the team EXCLUDING adding and removing members. This is to be done via add_member_to_team and remove_member_from_team
:param team: New instance of Team that is to be updated
:return: The modified Team instance
"""
disallowed_char = self._check_for_disallowed_char(team.name)
if disallowed_char:
raise NameNotAllowedError(disallowed_char)
disallowed_char = self._check_for_disallowed_char(team.abbreviation)
if disallowed_char:
raise NameNotAllowedError(disallowed_char)
if not team.name or len(team.name) > self.MAX_TEAM_NAME_LENGTH:
raise TeamNameTooLongError()
if not team.abbreviation or len(team.abbreviation) > self.MAX_TEAM_ABBR_LENGTH:
raise TeamAbbrInvalidError()
return await self._db_service.update_team(team)
async def add_member_to_team(self, team: Team, user: User, status: TeamStatus = TeamStatus.MEMBER) -> Team:
if user in team.members:
raise AlreadyMemberError()
await self._db_service.add_member_to_team(team, user, status)
return await self.get_team_by_id(team.id)
async def remove_member_from_team(self, team: Team, user: User) -> Team:
if user not in team.members:
raise NotMemberError()
if team.members[user] == TeamStatus.LEADER:
raise TeamLeadRemovalError()
await self._db_service.remove_user_from_team(team, user)
return await self.get_team_by_id(team.id)
async def is_join_password_valid(self, team_id: int, join_password: str) -> bool:
team = await self.get_team_by_id(team_id)
if not team:
return False
return team.join_password == join_password
def _check_for_disallowed_char(self, name: str) -> Optional[str]:
for c in name:
if c not in self.ALLOWED_TEAM_NAME_SYMBOLS:
return c
return None

View File

@ -0,0 +1,29 @@
from dataclasses import dataclass
from enum import Enum
from typing import Self
from src.ezgg_lan_manager.types.User import User
class TeamStatus(Enum):
MEMBER = 0
OFFICER = 1
LEADER = 2
@classmethod
def from_str(cls, team_status: str) -> Self:
if team_status == "MEMBER":
return TeamStatus.MEMBER
elif team_status == "OFFICER":
return TeamStatus.OFFICER
elif team_status == "LEADER":
return TeamStatus.LEADER
raise ValueError
@dataclass(frozen=True)
class Team:
id: int
name: str
abbreviation: str
members: dict[User, TeamStatus]
join_password: str

View File

@ -19,4 +19,9 @@ class User:
last_updated_at: datetime
def __hash__(self) -> int:
return hash(f"{self.user_id}{self.user_name}{self.user_mail}")
return hash(self.user_id)
def __eq__(self, other):
if not isinstance(other, User):
return NotImplemented
return self.user_id == other.user_id