74 lines
3.0 KiB
Python
74 lines
3.0 KiB
Python
import logging
|
|
from typing import Optional
|
|
|
|
from src.ez_lan_manager.services.AccountingService import AccountingService, InsufficientFundsError
|
|
from src.ez_lan_manager.services.DatabaseService import DatabaseService
|
|
from src.ez_lan_manager.types.ConfigurationTypes import LanInfo
|
|
from src.ez_lan_manager.types.Ticket import Ticket
|
|
|
|
logger = logging.getLogger(__name__.split(".")[-1])
|
|
|
|
class TicketNotAvailableError(Exception):
|
|
def __init__(self, category: str):
|
|
self.category = category
|
|
|
|
class UserAlreadyHasTicketError(Exception):
|
|
pass
|
|
|
|
class TicketingService:
|
|
def __init__(self, lan_info: LanInfo, db_service: DatabaseService, accounting_service: AccountingService) -> None:
|
|
self._lan_info = lan_info
|
|
self._db_service = db_service
|
|
self._accounting_service = accounting_service
|
|
|
|
def get_total_tickets(self) -> int:
|
|
return sum([self._lan_info.ticket_info.get_available_tickets(c) for c in self._lan_info.ticket_info.categories])
|
|
|
|
def get_available_tickets(self) -> dict[str, int]:
|
|
result = self._lan_info.ticket_info.total_available_tickets
|
|
all_tickets = self._db_service.get_tickets()
|
|
for ticket in all_tickets:
|
|
result[ticket.category] -= 1
|
|
|
|
return result
|
|
|
|
def purchase_ticket(self, user_id: int, category: str) -> Ticket:
|
|
if category not in self._lan_info.ticket_info.categories or self.get_available_tickets()[category] < 1:
|
|
raise TicketNotAvailableError(category)
|
|
|
|
user_balance = self._accounting_service.get_balance(user_id)
|
|
if self._lan_info.ticket_info.get_price(category) > user_balance:
|
|
raise InsufficientFundsError
|
|
|
|
if self.get_user_ticket(user_id):
|
|
raise UserAlreadyHasTicketError
|
|
|
|
if new_ticket := self._db_service.generate_ticket_for_user(user_id, category):
|
|
self._accounting_service.remove_balance(
|
|
user_id,
|
|
self._lan_info.ticket_info.get_price(new_ticket.category),
|
|
f"TICKET {new_ticket.ticket_id}"
|
|
)
|
|
logger.debug(f"User {user_id} purchased ticket {new_ticket.ticket_id}")
|
|
return new_ticket
|
|
|
|
raise RuntimeError("An unknown error occurred while purchasing ticket")
|
|
|
|
def refund_ticket(self, user_id: int) -> bool:
|
|
user_ticket = self.get_user_ticket(user_id)
|
|
if not user_ticket:
|
|
return False
|
|
|
|
if self._db_service.delete_ticket(user_ticket.ticket_id):
|
|
self._accounting_service.add_balance(user_id, self._lan_info.ticket_info.get_price(user_ticket.category), f"TICKET REFUND {user_ticket.ticket_id}")
|
|
logger.debug(f"User {user_id} refunded ticket {user_ticket.ticket_id}")
|
|
return True
|
|
|
|
return False
|
|
|
|
def transfer_ticket(self, ticket_id: int, user_id: int) -> bool:
|
|
return self._db_service.change_ticket_owner(ticket_id, user_id)
|
|
|
|
def get_user_ticket(self, user_id: int) -> Optional[Ticket]:
|
|
return self._db_service.get_ticket_for_user(user_id)
|