143 lines
4.7 KiB
Python
143 lines
4.7 KiB
Python
from datetime import datetime
|
|
import threading
|
|
from sqlalchemy.orm import mapped_column, relationship
|
|
from sqlalchemy.orm.attributes import Mapped
|
|
from sqlalchemy.orm.properties import ForeignKey
|
|
from sqlalchemy.schema import UniqueConstraint
|
|
from sqlalchemy.types import TIMESTAMP, BigInteger, Integer, String, Text
|
|
from sqlalchemy.sql import func
|
|
from sqlalchemy_utc import UtcDateTime
|
|
from discord_webhook import DiscordWebhook
|
|
import app_db
|
|
import spec
|
|
|
|
|
|
class Event(app_db.BaseModel):
|
|
__tablename__ = "events"
|
|
|
|
# surrogate key
|
|
id: Mapped[int] = mapped_column(Integer, autoincrement=True, primary_key=True)
|
|
|
|
# primary key
|
|
team_id: Mapped[int] = mapped_column(ForeignKey("teams.id"), nullable=False)
|
|
name: Mapped[str] = mapped_column(String(255), nullable=False)
|
|
start_time: Mapped[datetime] = mapped_column(UtcDateTime, nullable=False)
|
|
|
|
description: Mapped[str] = mapped_column(Text, nullable=True)
|
|
created_at: Mapped[datetime] = mapped_column(TIMESTAMP, server_default=func.now())
|
|
discord_message_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
|
|
|
|
team: Mapped["Team"] = relationship("Team", back_populates="events")
|
|
players: Mapped[list["PlayerEvent"]] = relationship("PlayerEvent", back_populates="event")
|
|
|
|
__table_args__ = (
|
|
UniqueConstraint("team_id", "name", "start_time"),
|
|
)
|
|
|
|
def get_discord_content(self):
|
|
start_timestamp = int(self.start_time.timestamp())
|
|
players = list(self.players)
|
|
# players with a role should be sorted first
|
|
players.sort(key=lambda p: p.role is not None, reverse=True)
|
|
players_info = []
|
|
|
|
for player in players:
|
|
player_info = "- "
|
|
|
|
if player.role:
|
|
player_info += f"**{player.role.role.name}:** "
|
|
|
|
player_info += f"{player.player.username}"
|
|
|
|
if player.has_confirmed:
|
|
player_info += " ✅"
|
|
else:
|
|
player_info += " ⏳"
|
|
|
|
players_info.append(player_info)
|
|
|
|
return "\n".join([
|
|
f"# {self.name}",
|
|
"",
|
|
self.description or "*No description.*",
|
|
"",
|
|
f"<t:{start_timestamp}:f>",
|
|
"\n".join(players_info),
|
|
"",
|
|
"[Confirm attendance here]" +
|
|
f"(https://availabili.tf/team/id/{self.team.id}/events/{self.id})",
|
|
])
|
|
|
|
def get_or_create_webhook(self):
|
|
integration = app_db.db.session.query(
|
|
TeamDiscordIntegration
|
|
).where(
|
|
TeamDiscordIntegration.team_id == self.team_id
|
|
).first()
|
|
|
|
if not integration:
|
|
return None
|
|
|
|
if self.discord_message_id:
|
|
return DiscordWebhook(
|
|
integration.webhook_url,
|
|
id=str(self.discord_message_id),
|
|
username=integration.webhook_bot_name,
|
|
avatar_url=integration.webhook_bot_profile_picture,
|
|
)
|
|
else:
|
|
return DiscordWebhook(
|
|
integration.webhook_url,
|
|
username=integration.webhook_bot_name,
|
|
avatar_url=integration.webhook_bot_profile_picture,
|
|
)
|
|
|
|
def update_discord_message(self):
|
|
webhook = self.get_or_create_webhook()
|
|
if webhook:
|
|
webhook.content = self.get_discord_content()
|
|
if webhook.id:
|
|
# fire and forget
|
|
threading.Thread(target=webhook.edit).start()
|
|
else:
|
|
webhook.execute()
|
|
if webhook_id := webhook.id:
|
|
self.discord_message_id = int(webhook_id)
|
|
app_db.db.session.commit()
|
|
else:
|
|
raise Exception("Failed to create webhook")
|
|
|
|
class EventSchema(spec.BaseModel):
|
|
id: int
|
|
team_id: int
|
|
name: str
|
|
description: str | None
|
|
start_time: datetime
|
|
created_at: datetime
|
|
|
|
@classmethod
|
|
def from_model(cls, model: Event) -> "EventSchema":
|
|
return cls(
|
|
id=model.id,
|
|
name=model.name,
|
|
description=model.description,
|
|
start_time=model.start_time,
|
|
team_id=model.team_id,
|
|
created_at=model.created_at,
|
|
)
|
|
|
|
class EventPlayersSchema(spec.BaseModel):
|
|
players: list["PlayerEventRolesSchema"]
|
|
|
|
@classmethod
|
|
def from_model(cls, model: Event) -> "EventPlayersSchema":
|
|
return cls(
|
|
players=[PlayerEventRolesSchema.from_model(player) for player in model.players],
|
|
roles=[RoleSchema.from_model(player.role.role) for player in model.players if player.role],
|
|
)
|
|
|
|
|
|
from models.team import Team
|
|
from models.player_event import PlayerEvent
|
|
from models.team_integration import TeamDiscordIntegration
|