Vereins-IT/ezgg-lan-manager
Archived
8
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add SeatingService with seating plan generation

Browse Source
This commit is contained in:
David Rodenkirchen
2024-08-20 15:34:36 +02:00
parent 96278258ef
commit b9b5e0ede0
10 changed files with 1487 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/EzLanManager.py
+3 -26
View File
@@ -1,28 +1,22 @@
import logging
from datetime import datetime, date
from from_root import from_root
from src.ez_lan_manager.services.AccountingService import AccountingService
from src.ez_lan_manager.services.ConfigurationService import ConfigurationService
from src.ez_lan_manager.services.DatabaseService import DatabaseService
from random import randint
from src.ez_lan_manager.services.MailingService import MailingService
from src.ez_lan_manager.services.NewsService import NewsService
from src.ez_lan_manager.services.SeatingService import SeatingService
from src.ez_lan_manager.services.TicketingService import TicketingService
from src.ez_lan_manager.services.UserService import UserService
from src.ez_lan_manager.types.News import News
from src.ez_lan_manager.types.Transaction import Transaction
from src.ez_lan_manager.types.User import User
logger = logging.getLogger(__name__.split(".")[-1])
if __name__ == "__main__":
logging.basicConfig(level=logging.DEBUG)
configuration_service = ConfigurationService(from_root("config.toml"))
lan_info = configuration_service.get_lan_info()
seating_config = configuration_service.get_seating_configuration()
db_config = configuration_service.get_database_configuration()
db_service = DatabaseService(db_config)
user_service = UserService(db_service)
@@ -30,21 +24,4 @@ if __name__ == "__main__":
news_service = NewsService(db_service)
mailing_service = MailingService(configuration_service.get_mailing_service_configuration())
ticketing_service = TicketingService(lan_info, db_service, accounting_service)
print(ticketing_service.refund_ticket(19))
#print(ticketing_service.get_available_tickets())
#user_service.create_user("Alex", "alex@gmail.com", "MeinPasswort")
#print(user_service.is_login_valid("Alex@gmail.com", "MeinPasswort"))
# news_service.add_news(News(
# news_id=None,
# title=f"TITLE{randint(0, 9999)}",
# subtitle="",
# content="",
# author=user_service.get_user(19),
# news_date=date(2024, 8, 30)
# ))
seating_service = SeatingService(seating_config, lan_info, db_service, ticketing_service)
src/ez_lan_manager/services/ConfigurationService.py
+18 -1
View File
@@ -4,7 +4,9 @@ from pathlib import Path
import logging
import tomllib
from src.ez_lan_manager.types.ConfigurationTypes import DatabaseConfiguration, MailingServiceConfiguration, LanInfo, TicketInfo
from from_root import from_root
from src.ez_lan_manager.types.ConfigurationTypes import DatabaseConfiguration, MailingServiceConfiguration, LanInfo, TicketInfo, SeatingConfiguration
logger = logging.getLogger(__name__.split(".")[-1])
@@ -50,6 +52,7 @@ class ConfigurationService:
try:
lan_info = self._config["lan"]
ticket_info = TicketInfo(
default_category=lan_info["default_category"],
categories=list(lan_info["tickets"].keys()),
_prices=lan_info["prices"],
_available_tickets=lan_info["tickets"]
@@ -64,3 +67,17 @@ class ConfigurationService:
except KeyError:
logger.fatal("Error loading LAN Info, exiting...")
sys.exit(1)
def get_seating_configuration(self) -> SeatingConfiguration:
try:
seating_config = self._config["seating"]
base_svg_file_path = from_root(seating_config["base_svg_path"])
if not base_svg_file_path.exists():
logger.fatal(f"Specified seating plan SVG file was not found at {base_svg_file_path}! Exiting...")
sys.exit(1)
return SeatingConfiguration(
base_svg_path=base_svg_file_path
)
except KeyError:
logger.fatal("Error loading seating configuration, exiting...")
sys.exit(1)
src/ez_lan_manager/services/DatabaseService.py
+43
View File
@@ -8,6 +8,7 @@ from mariadb import Cursor
from src.ez_lan_manager.types.ConfigurationTypes import DatabaseConfiguration
from src.ez_lan_manager.types.News import News
from src.ez_lan_manager.types.Seat import Seat
from src.ez_lan_manager.types.Ticket import Ticket
from src.ez_lan_manager.types.Transaction import Transaction
from src.ez_lan_manager.types.User import User
@@ -251,3 +252,45 @@ class DatabaseService:
logger.warning(f"Error deleting ticket: {e}")
return False
return True
def generate_fresh_seats_table(self, seats: list[tuple[str, str]]) -> None:
""" WARNING: THIS WILL DELETE ALL EXISTING DATA! DO NOT USE ON PRODUCTION DATABASE! """
cursor = self._get_cursor()
try:
cursor.execute("TRUNCATE seats;")
for seat in seats:
cursor.execute("INSERT INTO seats (seat_id, seat_category) VALUES (?, ?);", (seat[0], seat[1]))
self._connection.commit()
except Exception as e:
logger.warning(f"Error generating fresh seats table: {e}")
return
def get_seating_info(self) -> list[Seat]:
results = []
cursor = self._get_cursor()
try:
cursor.execute("SELECT seats.*, users.* FROM seats LEFT JOIN users ON seats.user = users.user_id;")
except Exception as e:
logger.warning(f"Error getting seats table: {e}")
return results
for seat_raw in cursor.fetchall():
if seat_raw[3] is None: # Empty seat
results.append(Seat(seat_raw[0], bool(seat_raw[1]), seat_raw[2], None))
else:
user = self._map_db_result_to_user(seat_raw[4:])
results.append(Seat(seat_raw[0], bool(seat_raw[1]), seat_raw[2], user))
return results
def seat_user(self, seat_id: str, user_id: int) -> bool:
cursor = self._get_cursor()
try:
cursor.execute("UPDATE seats SET user = ? WHERE seat_id = ?;", (user_id, seat_id))
affected_rows = cursor.rowcount
self._connection.commit()
except Exception as e:
logger.warning(f"Error seating user: {e}")
return False
return bool(affected_rows)
src/ez_lan_manager/services/SeatingService.py
+152
View File
@@ -0,0 +1,152 @@
import logging
import re
from io import StringIO
from pathlib import Path
from typing import Optional
from xml.etree import ElementTree
from from_root import from_root
from src.ez_lan_manager.services.DatabaseService import DatabaseService
from src.ez_lan_manager.services.TicketingService import TicketingService
from src.ez_lan_manager.types.ConfigurationTypes import LanInfo, SeatingConfiguration
from src.ez_lan_manager.types.Seat import Seat
logger = logging.getLogger(__name__.split(".")[-1])
class NoTicketError(Exception):
pass
class SeatNotFoundError(Exception):
pass
class WrongCategoryError(Exception):
pass
class SeatAlreadyTakenError(Exception):
pass
class SeatingService:
def __init__(self, seating_configuration: SeatingConfiguration, lan_info: LanInfo, db_service: DatabaseService, ticketing_service: TicketingService) -> None:
self._seating_configuration = seating_configuration
self._lan_info = lan_info
self._db_service = db_service
self._ticketing_service = ticketing_service
self._seating_plan = StringIO()
ElementTree.parse(self._seating_configuration.base_svg_path).write(self._seating_plan, encoding="unicode")
def get_seating(self) -> list[Seat]:
return self._db_service.get_seating_info()
def get_seat(self, seat_id: str, cached_data: Optional[list[Seat]] = None) -> Optional[Seat]:
all_seats = self.get_seating() if not cached_data else cached_data
for seat in all_seats:
if seat.seat_id == seat_id:
return seat
def seat_user(self, user_id: int, seat_id: str) -> None:
user_ticket = self._ticketing_service.get_user_ticket(user_id)
if not user_ticket:
raise NoTicketError
seat = self.get_seat(seat_id)
if not seat:
raise SeatNotFoundError
if seat.category != user_ticket.category:
raise WrongCategoryError
if seat.user is not None:
raise SeatAlreadyTakenError
self._db_service.seat_user(seat_id, user_id)
self.update_svg_with_seating_status()
def generate_new_seating_table(self, seating_plan_fp: Path, no_confirm: bool = False) -> None:
if not no_confirm:
confirm = input("WARNING: THIS ACTION WILL DELETE ALL SEATING DATA! TYPE 'AGREE' TO CONTINUE: ")
if confirm != "AGREE":
logging.info("Seating table generation aborted...")
return
et = ElementTree.parse(seating_plan_fp)
seat_ids = []
for child in et.getroot().findall(".//mxCell"):
possible_seat_identifier = child.get("value")
try:
if re.match(r"^\w\d{1,3}$", possible_seat_identifier):
seat_ids.append((possible_seat_identifier, self._lan_info.ticket_info.default_category))
except TypeError:
continue
for child in et.getroot().findall(".//object"):
possible_seat_identifier = child.get("label")
try:
if re.match(r"^\w\d{1,3}$", possible_seat_identifier):
category = child.get("category")
seat_ids.append((possible_seat_identifier, category))
except TypeError:
continue
self._db_service.generate_fresh_seats_table(sorted(seat_ids, key=lambda sd: sd[0]))
self.update_svg_with_seating_status()
def update_svg_with_seating_status(self) -> None:
et = ElementTree.parse(self._seating_configuration.base_svg_path)
root = et.getroot()
namespace = {'svg': root.tag.split('}')[0].strip('{')} if '}' in root.tag else {}
rect_g_pairs = []
last_rect = None
for elem in root.iter():
if elem.tag == f"{{{namespace.get('svg')}}}rect":
last_rect = elem
elif elem.tag == f"{{{namespace.get('svg')}}}g":
if last_rect is not None:
rect_g_pairs.append((last_rect, elem))
last_rect = None
all_seats = self.get_seating()
for rect, g in rect_g_pairs:
seat_id = self.get_seat_id_from_element(g, namespace)
if not seat_id:
continue
seat = self.get_seat(seat_id, cached_data=all_seats)
if not seat.is_blocked and seat.user is None:
rect.set("fill", "rgb(102, 255, 51)")
elif not seat.is_blocked and seat.user is not None:
rect.set("fill", "rgb(204, 0, 0)")
else:
rect.set("fill", "rgb(190,190,190)")
# @ToDo: Set URL's properly
rect.set('onclick', f"window.open('https://httpbin.org/get?seat_id={seat_id}', '_blank')")
g.set('onclick', f"window.open('https://httpbin.org/get?seat_id={seat_id}', '_blank')")
# Debug output
et.write(from_root("debug_seating_plan.svg"))
self._seating_plan = StringIO()
et.write(self._seating_plan, encoding='unicode')
@staticmethod
def get_seat_id_from_element(element: ElementTree.Element, namespace: dict) -> Optional[str]:
seat_id = None
for child in element.iter():
if child.tag == f"{{{namespace.get('svg')}}}text":
# Extract identifier from <text> element
seat_id = child.text.strip() if child.text else None
elif child.tag.endswith('div') and child.text:
# Extract identifier from <foreignObject>/<div>
seat_id = child.text.strip()
if seat_id: # Break if we've already found the identifier
break
try:
if re.match(r"^\w\d{1,3}$", seat_id):
return seat_id
except TypeError:
pass
return
src/ez_lan_manager/types/ConfigurationTypes.py
+7
View File
@@ -1,6 +1,8 @@
from copy import copy
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
class NoSuchCategoryError(Exception):
pass
@@ -15,6 +17,7 @@ class DatabaseConfiguration:
@dataclass(frozen=True)
class TicketInfo:
default_category: str
categories: list[str]
_prices: dict[str, int]
_available_tickets: dict[str, int]
@@ -50,3 +53,7 @@ class LanInfo:
ticket_info: TicketInfo
date_from: datetime
date_till: datetime
@dataclass(frozen=True)
class SeatingConfiguration:
base_svg_path: Path
src/ez_lan_manager/types/Seat.py
+12
View File
@@ -0,0 +1,12 @@
from dataclasses import dataclass
from typing import Optional
from src.ez_lan_manager.types.User import User
@dataclass(frozen=True)
class Seat:
seat_id: str
is_blocked: bool
category: str
user: Optional[User]