profitroll/QuizBot
1
0
Fork 0
Code Issues 5 Pull Requests Actions Projects 1 Releases 2 Activity

Implemented #7

Browse Source
This commit is contained in:
Profitroll
2025-04-25 11:37:44 +02:00
parent d9f563285c
commit fb16fa2bc8
9 changed files with 175 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
classes/pycord_bot.py
View File

@@ -1,10 +1,11 @@
from datetime import datetime
from logging import Logger
from pathlib import Path
from typing import Any, override, List, Dict
from zoneinfo import ZoneInfo
from bson import ObjectId
from discord import Guild, User, TextChannel
from discord import Guild, User, TextChannel, Attachment, File
from libbot.cache.classes import CacheMemcached, CacheRedis
from libbot.cache.manager import create_cache_client
from libbot.pycord.classes import PycordBot as LibPycordBot
@@ -59,7 +60,6 @@ class PycordBot(LibPycordBot):
async def _execute_event_controller(self) -> None:
await self._process_events_start()
await self._process_events_end()
# await self._process_events_post_end()
async def _process_events_start(self) -> None:
# Get events to start
@@ -69,22 +69,45 @@ class PycordBot(LibPycordBot):
# Process each event
for event in events:
if len(event.stage_ids) == 0:
# TODO Make a nice message for management
logger.error("Could not start the event %s: no event stages are defined.", event._id)
continue
guild: Guild = self.get_guild(event.guild_id)
pycord_guild: PycordGuild = await self.find_guild(guild)
first_stage: PycordEventStage = await self.find_event_stage(event.stage_ids[0])
# Get list of participants
users: List[PycordUser] = await self._get_event_participants(event._id)
for user in users:
await user._set(self.cache, current_event_id=event._id, current_stage_id=first_stage._id)
# Create a channel for each participant
await user.setup_event_channel(self, guild, pycord_guild, event, cache=self.cache)
# Send a notification about event start
user_channel: TextChannel = guild.get_channel(user.event_channels[str(event._id)])
thumbnail: File | None = (
None
if event.thumbnail is None
else File(Path(f"data/{event.thumbnail['id']}"), event.thumbnail["filename"])
)
# TODO Make a nice message
# TODO Also send a thumbnail, event info and short explanation on how to play
await user_channel.send(f"Event **{event.name}** is starting!")
await user_channel.send(
f"Event **{event.name}** is starting!\n\nUse slash command `/guess` to suggest your answers to each event stage.",
file=thumbnail,
)
first_stage_files: List[File] | None = first_stage.get_media_files()
await user_channel.send(
f"First stage...\n\n{first_stage.question}", files=first_stage_files
)
# TODO Make a nice message
await self._notify_admins(
@@ -103,6 +126,13 @@ class PycordBot(LibPycordBot):
for event in events:
guild: Guild = self.get_guild(event.guild_id)
pycord_guild: PycordGuild = await self.find_guild(guild)
stages: List[PycordEventStage] = await self._get_event_stages(event)
# TODO Make a nice message
stages_string: str = "\n\n".join(
f"**Stage {stage.sequence+1}**\nQuestion: {stage.question}\nAnswer: ||{stage.answer}||"
for stage in stages
)
# Get list of participants
users: List[PycordUser] = await self._get_event_participants(event._id)
@@ -112,8 +142,9 @@ class PycordBot(LibPycordBot):
user_channel: TextChannel = guild.get_channel(user.event_channels[str(event._id)])
# TODO Make a nice message
# TODO Reveal answers to stages
await user_channel.send(f"Event **{event.name}** has ended!")
await user_channel.send(
f"Event **{event.name}** has ended! Stages and respective answers are listed below.\n\n{stages_string}"
)
# Lock each participant out
await user.lock_event_channel(guild, event._id, channel=user_channel)
@@ -125,37 +156,6 @@ class PycordBot(LibPycordBot):
f"Event **{event.name}** has ended! Users can no longer submit their answers.",
)
# async def _process_events_post_end(self) -> None:
# # Get events that ended an hour ago
# # TODO Replace with 1 hour after testing!
# events: List[PycordEvent] = await self._get_events(
# {
# "ends": datetime.now(tz=ZoneInfo("UTC")).replace(second=0, microsecond=0)
# - timedelta(minutes=1)
# }
# )
#
# # Process each event
# for event in events:
# guild: Guild = self.get_guild(event.guild_id)
# pycord_guild: PycordGuild = await self.find_guild(guild)
#
# # Get list of participants
# users: List[PycordUser] = await self._get_event_participants(event._id)
#
# for user in users:
# # Send a notification about event start
# user_channel: TextChannel = guild.get_channel(user.event_channels[str(event._id)])
#
# # Remove their view permissions
# await user.lock_event_channel(guild, event._id, completely=True, channel=user_channel)
#
# await self._notify_admins(
# guild,
# pycord_guild,
# f"Access has been updated, users can no longer access their channels for the event **{event.name}**.",
# )
@staticmethod
async def _get_events(query: Dict[str, Any]) -> List[PycordEvent]:
events: List[PycordEvent] = []
@@ -174,6 +174,9 @@ class PycordBot(LibPycordBot):
return users
async def _get_event_stages(self, event: PycordEvent) -> List[PycordEventStage]:
return [(await self.find_event_stage(stage_id)) for stage_id in event.stage_ids]
@staticmethod
async def _notify_admins(guild: Guild, pycord_guild: PycordGuild, message: str) -> None:
management_channel: TextChannel | None = guild.get_channel(pycord_guild.channel_id)
@@ -264,3 +267,13 @@ class PycordBot(LibPycordBot):
# TODO Document this method
async def find_event_stage(self, stage_id: str | ObjectId) -> PycordEventStage:
return await PycordEventStage.from_id(stage_id, cache=self.cache)
@staticmethod
async def process_attachments(attachments: List[Attachment]) -> List[Dict[str, Any]]:
processed_attachments: List[Dict[str, Any]] = []
for attachment in attachments:
await attachment.save(Path(f"data/{attachment.id}"))
processed_attachments.append({"id": attachment.id, "filename": attachment.filename})
return processed_attachments

12
classes/pycord_event.py
View File

@@ -27,7 +27,7 @@ class PycordEvent:
"creator_id",
"starts",
"ends",
"thumbnail_id",
"thumbnail",
"stage_ids",
)
__short_name__ = "event"
@@ -42,7 +42,7 @@ class PycordEvent:
creator_id: int
starts: datetime
ends: datetime
thumbnail_id: str | None
thumbnail: Dict[str, Any] | None
stage_ids: List[ObjectId]
@classmethod
@@ -105,7 +105,7 @@ class PycordEvent:
creator_id: int,
starts: datetime,
ends: datetime,
thumbnail_id: str | None,
thumbnail: Dict[str, Any] | None,
cache: Optional[Cache] = None,
) -> "PycordEvent":
db_entry: Dict[str, Any] = {
@@ -117,7 +117,7 @@ class PycordEvent:
"creator_id": creator_id,
"starts": starts,
"ends": ends,
"thumbnail_id": thumbnail_id,
"thumbnail": thumbnail,
"stage_ids": [],
}
@@ -213,7 +213,7 @@ class PycordEvent:
"creator_id": self.creator_id,
"starts": self.starts,
"ends": self.ends,
"thumbnail_id": self.thumbnail_id,
"thumbnail": self.thumbnail,
"stage_ids": self.stage_ids,
}
@@ -228,7 +228,7 @@ class PycordEvent:
"creator_id": None,
"starts": None,
"ends": None,
"thumbnail_id": None,
"thumbnail": None,
"stage_ids": [],
}

12
classes/pycord_event_stage.py
View File

@@ -1,10 +1,12 @@
from dataclasses import dataclass
from datetime import datetime
from logging import Logger
from pathlib import Path
from typing import Any, Dict, List, Optional
from zoneinfo import ZoneInfo
from bson import ObjectId
from discord import File
from libbot.cache.classes import Cache
from pymongo.results import InsertOneResult
@@ -38,7 +40,7 @@ class PycordEventStage:
creator_id: int
question: str
answer: str
media: List[int]
media: List[Dict[str, Any]]
@classmethod
async def from_id(cls, stage_id: str | ObjectId, cache: Optional[Cache] = None) -> "PycordEventStage":
@@ -235,3 +237,11 @@ class PycordEventStage:
"""
await self.__collection__.delete_one({"_id": self._id})
self._delete_cache(cache)
# TODO Add documentation
def get_media_files(self) -> List[File] | None:
return (
None
if len(self.media) == 0
else [File(Path(f"data/{media['id']}"), media["filename"]) for media in self.media]
)

6
classes/pycord_user.py
View File

@@ -361,3 +361,9 @@ class PycordUser:
self.event_channels[event_id if isinstance(event_id, str) else str(event_id)] = channel_id
await self._set(cache, event_channels=self.event_channels)
# TODO Add documentation
async def set_event_stage(self, stage_id: str | ObjectId | None, cache: Optional[Cache] = None) -> None:
await self._set(
cache, current_stage_id=stage_id if isinstance(stage_id, str) else ObjectId(stage_id)
)