mirror of https://github.com/stijndcl/didier
Remove wordle
parent
c3a7ff8e4c
commit
4b25d5d519
|
@ -1,2 +0,0 @@
|
||||||
WORDLE_GUESS_COUNT = 6
|
|
||||||
WORDLE_WORD_LENGTH = 5
|
|
|
@ -1,85 +0,0 @@
|
||||||
import datetime
|
|
||||||
from typing import Optional
|
|
||||||
|
|
||||||
from sqlalchemy import delete, select
|
|
||||||
from sqlalchemy.ext.asyncio import AsyncSession
|
|
||||||
|
|
||||||
from database.crud.users import get_or_add_user
|
|
||||||
from database.schemas import WordleGuess, WordleWord
|
|
||||||
|
|
||||||
__all__ = [
|
|
||||||
"get_active_wordle_game",
|
|
||||||
"get_wordle_guesses",
|
|
||||||
"make_wordle_guess",
|
|
||||||
"set_daily_word",
|
|
||||||
"reset_wordle_games",
|
|
||||||
]
|
|
||||||
|
|
||||||
|
|
||||||
async def get_active_wordle_game(session: AsyncSession, user_id: int) -> list[WordleGuess]:
|
|
||||||
"""Find a player's active game"""
|
|
||||||
await get_or_add_user(session, user_id)
|
|
||||||
statement = select(WordleGuess).where(WordleGuess.user_id == user_id)
|
|
||||||
guesses = (await session.execute(statement)).scalars().all()
|
|
||||||
return guesses
|
|
||||||
|
|
||||||
|
|
||||||
async def get_wordle_guesses(session: AsyncSession, user_id: int) -> list[str]:
|
|
||||||
"""Get the strings of a player's guesses"""
|
|
||||||
active_game = await get_active_wordle_game(session, user_id)
|
|
||||||
return list(map(lambda g: g.guess.lower(), active_game))
|
|
||||||
|
|
||||||
|
|
||||||
async def make_wordle_guess(session: AsyncSession, user_id: int, guess: str):
|
|
||||||
"""Make a guess in your current game"""
|
|
||||||
guess_instance = WordleGuess(user_id=user_id, guess=guess)
|
|
||||||
session.add(guess_instance)
|
|
||||||
await session.commit()
|
|
||||||
|
|
||||||
|
|
||||||
async def get_daily_word(session: AsyncSession) -> Optional[WordleWord]:
|
|
||||||
"""Get the word of today"""
|
|
||||||
statement = select(WordleWord).where(WordleWord.day == datetime.date.today())
|
|
||||||
row = (await session.execute(statement)).scalar_one_or_none()
|
|
||||||
|
|
||||||
if row is None:
|
|
||||||
return None
|
|
||||||
|
|
||||||
return row
|
|
||||||
|
|
||||||
|
|
||||||
async def set_daily_word(session: AsyncSession, word: str, *, forced: bool = False) -> str:
|
|
||||||
"""Set the word of today
|
|
||||||
|
|
||||||
This does NOT overwrite the existing word if there is one, so that it can safely run
|
|
||||||
on startup every time.
|
|
||||||
|
|
||||||
In order to always overwrite the current word, set the "forced"-kwarg to True.
|
|
||||||
|
|
||||||
Returns the word that was chosen. If one already existed, return that instead.
|
|
||||||
"""
|
|
||||||
current_word = await get_daily_word(session)
|
|
||||||
|
|
||||||
if current_word is None:
|
|
||||||
current_word = WordleWord(word=word, day=datetime.date.today())
|
|
||||||
session.add(current_word)
|
|
||||||
await session.commit()
|
|
||||||
|
|
||||||
# Remove all active games
|
|
||||||
await reset_wordle_games(session)
|
|
||||||
elif forced:
|
|
||||||
current_word.word = word
|
|
||||||
session.add(current_word)
|
|
||||||
await session.commit()
|
|
||||||
|
|
||||||
# Remove all active games
|
|
||||||
await reset_wordle_games(session)
|
|
||||||
|
|
||||||
return current_word.word
|
|
||||||
|
|
||||||
|
|
||||||
async def reset_wordle_games(session: AsyncSession):
|
|
||||||
"""Reset all active games"""
|
|
||||||
statement = delete(WordleGuess)
|
|
||||||
await session.execute(statement)
|
|
||||||
await session.commit()
|
|
|
@ -1,60 +0,0 @@
|
||||||
from datetime import date
|
|
||||||
|
|
||||||
from sqlalchemy import select
|
|
||||||
from sqlalchemy.ext.asyncio import AsyncSession
|
|
||||||
|
|
||||||
from database.crud.users import get_or_add_user
|
|
||||||
from database.schemas import WordleStats
|
|
||||||
|
|
||||||
__all__ = ["get_wordle_stats", "complete_wordle_game"]
|
|
||||||
|
|
||||||
|
|
||||||
async def get_wordle_stats(session: AsyncSession, user_id: int) -> WordleStats:
|
|
||||||
"""Get a user's wordle stats
|
|
||||||
|
|
||||||
If no entry is found, it is first created
|
|
||||||
"""
|
|
||||||
await get_or_add_user(session, user_id)
|
|
||||||
|
|
||||||
statement = select(WordleStats).where(WordleStats.user_id == user_id)
|
|
||||||
stats = (await session.execute(statement)).scalar_one_or_none()
|
|
||||||
if stats is not None:
|
|
||||||
return stats
|
|
||||||
|
|
||||||
stats = WordleStats(user_id=user_id)
|
|
||||||
session.add(stats)
|
|
||||||
await session.commit()
|
|
||||||
await session.refresh(stats)
|
|
||||||
|
|
||||||
return stats
|
|
||||||
|
|
||||||
|
|
||||||
async def complete_wordle_game(session: AsyncSession, user_id: int, win: bool):
|
|
||||||
"""Update the user's Wordle stats"""
|
|
||||||
stats = await get_wordle_stats(session, user_id)
|
|
||||||
stats.games += 1
|
|
||||||
|
|
||||||
if win:
|
|
||||||
stats.wins += 1
|
|
||||||
|
|
||||||
# Update streak
|
|
||||||
today = date.today()
|
|
||||||
last_win = stats.last_win
|
|
||||||
stats.last_win = today
|
|
||||||
|
|
||||||
if last_win is None or (today - last_win).days > 1:
|
|
||||||
# Never won a game before or streak is over
|
|
||||||
stats.current_streak = 1
|
|
||||||
else:
|
|
||||||
# On a streak: increase counter
|
|
||||||
stats.current_streak += 1
|
|
||||||
|
|
||||||
# Update max streak if necessary
|
|
||||||
if stats.current_streak > stats.highest_streak:
|
|
||||||
stats.highest_streak = stats.current_streak
|
|
||||||
else:
|
|
||||||
# Streak is over
|
|
||||||
stats.current_streak = 0
|
|
||||||
|
|
||||||
session.add(stats)
|
|
||||||
await session.commit()
|
|
|
@ -45,9 +45,6 @@ __all__ = [
|
||||||
"UforaCourse",
|
"UforaCourse",
|
||||||
"UforaCourseAlias",
|
"UforaCourseAlias",
|
||||||
"User",
|
"User",
|
||||||
"WordleGuess",
|
|
||||||
"WordleStats",
|
|
||||||
"WordleWord",
|
|
||||||
]
|
]
|
||||||
|
|
||||||
|
|
||||||
|
@ -343,47 +340,3 @@ class User(Base):
|
||||||
reminders: list[Reminder] = relationship(
|
reminders: list[Reminder] = relationship(
|
||||||
"Reminder", back_populates="user", uselist=True, lazy="selectin", cascade="all, delete-orphan"
|
"Reminder", back_populates="user", uselist=True, lazy="selectin", cascade="all, delete-orphan"
|
||||||
)
|
)
|
||||||
wordle_guesses: list[WordleGuess] = relationship(
|
|
||||||
"WordleGuess", back_populates="user", uselist=True, lazy="selectin", cascade="all, delete-orphan"
|
|
||||||
)
|
|
||||||
wordle_stats: WordleStats = relationship(
|
|
||||||
"WordleStats", back_populates="user", uselist=False, lazy="selectin", cascade="all, delete-orphan"
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
class WordleGuess(Base):
|
|
||||||
"""A user's Wordle guesses for today"""
|
|
||||||
|
|
||||||
__tablename__ = "wordle_guesses"
|
|
||||||
|
|
||||||
wordle_guess_id: int = Column(Integer, primary_key=True)
|
|
||||||
user_id: int = Column(BigInteger, ForeignKey("users.user_id"))
|
|
||||||
guess: str = Column(Text, nullable=False)
|
|
||||||
|
|
||||||
user: User = relationship("User", back_populates="wordle_guesses", uselist=False, lazy="selectin")
|
|
||||||
|
|
||||||
|
|
||||||
class WordleStats(Base):
|
|
||||||
"""Stats about a user's wordle performance"""
|
|
||||||
|
|
||||||
__tablename__ = "wordle_stats"
|
|
||||||
|
|
||||||
wordle_stats_id: int = Column(Integer, primary_key=True)
|
|
||||||
user_id: int = Column(BigInteger, ForeignKey("users.user_id"))
|
|
||||||
last_win: Optional[date] = Column(Date, nullable=True)
|
|
||||||
games: int = Column(Integer, server_default="0", nullable=False)
|
|
||||||
wins: int = Column(Integer, server_default="0", nullable=False)
|
|
||||||
current_streak: int = Column(Integer, server_default="0", nullable=False)
|
|
||||||
highest_streak: int = Column(Integer, server_default="0", nullable=False)
|
|
||||||
|
|
||||||
user: User = relationship("User", back_populates="wordle_stats", uselist=False, lazy="selectin")
|
|
||||||
|
|
||||||
|
|
||||||
class WordleWord(Base):
|
|
||||||
"""The current Wordle word"""
|
|
||||||
|
|
||||||
__tablename__ = "wordle_word"
|
|
||||||
|
|
||||||
word_id: int = Column(Integer, primary_key=True)
|
|
||||||
word: str = Column(Text, nullable=False)
|
|
||||||
day: date = Column(Date, nullable=False, unique=True)
|
|
||||||
|
|
|
@ -4,8 +4,8 @@ from discord import app_commands
|
||||||
from overrides import overrides
|
from overrides import overrides
|
||||||
from sqlalchemy.ext.asyncio import AsyncSession
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||||||
|
|
||||||
from database.crud import easter_eggs, links, memes, ufora_courses, wordle
|
from database.crud import easter_eggs, links, memes, ufora_courses
|
||||||
from database.schemas import EasterEgg, WordleWord
|
from database.schemas import EasterEgg
|
||||||
|
|
||||||
__all__ = ["CacheManager", "EasterEggCache", "LinkCache", "UforaCourseCache"]
|
__all__ = ["CacheManager", "EasterEggCache", "LinkCache", "UforaCourseCache"]
|
||||||
|
|
||||||
|
@ -132,17 +132,6 @@ class UforaCourseCache(DatabaseCache):
|
||||||
return [app_commands.Choice(name=suggestion, value=suggestion.lower()) for suggestion in suggestions]
|
return [app_commands.Choice(name=suggestion, value=suggestion.lower()) for suggestion in suggestions]
|
||||||
|
|
||||||
|
|
||||||
class WordleCache(DatabaseCache):
|
|
||||||
"""Cache to store the current daily Wordle word"""
|
|
||||||
|
|
||||||
word: WordleWord
|
|
||||||
|
|
||||||
async def invalidate(self, database_session: AsyncSession):
|
|
||||||
word = await wordle.get_daily_word(database_session)
|
|
||||||
if word is not None:
|
|
||||||
self.word = word
|
|
||||||
|
|
||||||
|
|
||||||
class CacheManager:
|
class CacheManager:
|
||||||
"""Class that keeps track of all caches"""
|
"""Class that keeps track of all caches"""
|
||||||
|
|
||||||
|
@ -150,14 +139,12 @@ class CacheManager:
|
||||||
links: LinkCache
|
links: LinkCache
|
||||||
memes: MemeCache
|
memes: MemeCache
|
||||||
ufora_courses: UforaCourseCache
|
ufora_courses: UforaCourseCache
|
||||||
wordle_word: WordleCache
|
|
||||||
|
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
self.easter_eggs = EasterEggCache()
|
self.easter_eggs = EasterEggCache()
|
||||||
self.links = LinkCache()
|
self.links = LinkCache()
|
||||||
self.memes = MemeCache()
|
self.memes = MemeCache()
|
||||||
self.ufora_courses = UforaCourseCache()
|
self.ufora_courses = UforaCourseCache()
|
||||||
self.wordle_word = WordleCache()
|
|
||||||
|
|
||||||
async def initialize_caches(self, postgres_session: AsyncSession):
|
async def initialize_caches(self, postgres_session: AsyncSession):
|
||||||
"""Initialize the contents of all caches"""
|
"""Initialize the contents of all caches"""
|
||||||
|
@ -165,4 +152,3 @@ class CacheManager:
|
||||||
await self.links.invalidate(postgres_session)
|
await self.links.invalidate(postgres_session)
|
||||||
await self.memes.invalidate(postgres_session)
|
await self.memes.invalidate(postgres_session)
|
||||||
await self.ufora_courses.invalidate(postgres_session)
|
await self.ufora_courses.invalidate(postgres_session)
|
||||||
await self.wordle_word.invalidate(postgres_session)
|
|
||||||
|
|
|
@ -1,14 +1,6 @@
|
||||||
from typing import Optional
|
|
||||||
|
|
||||||
import discord
|
|
||||||
from discord import app_commands
|
|
||||||
from discord.ext import commands
|
from discord.ext import commands
|
||||||
|
|
||||||
from database.constants import WORDLE_WORD_LENGTH
|
|
||||||
from database.crud.wordle import get_wordle_guesses, make_wordle_guess
|
|
||||||
from database.crud.wordle_stats import complete_wordle_game
|
|
||||||
from didier import Didier
|
from didier import Didier
|
||||||
from didier.data.embeds.wordle import WordleEmbed, WordleErrorEmbed, is_wordle_game_over
|
|
||||||
|
|
||||||
|
|
||||||
class Games(commands.Cog):
|
class Games(commands.Cog):
|
||||||
|
@ -19,53 +11,6 @@ class Games(commands.Cog):
|
||||||
def __init__(self, client: Didier):
|
def __init__(self, client: Didier):
|
||||||
self.client = client
|
self.client = client
|
||||||
|
|
||||||
@app_commands.command(name="wordle", description="Play Wordle!")
|
|
||||||
async def wordle(self, interaction: discord.Interaction, guess: Optional[str] = None):
|
|
||||||
"""View your active Wordle game
|
|
||||||
|
|
||||||
If an argument is provided, make a guess instead
|
|
||||||
"""
|
|
||||||
await interaction.response.defer(ephemeral=True)
|
|
||||||
|
|
||||||
# Guess is wrong length
|
|
||||||
if guess is not None and len(guess) != 0 and len(guess) != WORDLE_WORD_LENGTH:
|
|
||||||
embed = WordleErrorEmbed(message=f"Guess must be 5 characters, but `{guess}` is {len(guess)}.").to_embed()
|
|
||||||
return await interaction.followup.send(embed=embed)
|
|
||||||
|
|
||||||
word_instance = self.client.database_caches.wordle_word.word
|
|
||||||
|
|
||||||
async with self.client.postgres_session as session:
|
|
||||||
guesses = await get_wordle_guesses(session, interaction.user.id)
|
|
||||||
|
|
||||||
# Trying to guess with a complete game
|
|
||||||
if is_wordle_game_over(guesses, word_instance.word):
|
|
||||||
embed = WordleErrorEmbed(
|
|
||||||
message="You've already completed today's Wordle.\nTry again tomorrow!"
|
|
||||||
).to_embed()
|
|
||||||
return await interaction.followup.send(embed=embed)
|
|
||||||
|
|
||||||
# Make a guess
|
|
||||||
if guess:
|
|
||||||
# The guess is not a real word
|
|
||||||
if guess.lower() not in self.client.wordle_words:
|
|
||||||
embed = WordleErrorEmbed(message=f"`{guess}` is not a valid word.").to_embed()
|
|
||||||
return await interaction.followup.send(embed=embed)
|
|
||||||
|
|
||||||
guess = guess.lower()
|
|
||||||
await make_wordle_guess(session, interaction.user.id, guess)
|
|
||||||
|
|
||||||
# Don't re-request the game, we already have it
|
|
||||||
# just append locally
|
|
||||||
guesses.append(guess)
|
|
||||||
|
|
||||||
embed = WordleEmbed(guesses=guesses, word=word_instance).to_embed()
|
|
||||||
await interaction.followup.send(embed=embed)
|
|
||||||
|
|
||||||
# After responding to the interaction: update stats in the background
|
|
||||||
game_over = is_wordle_game_over(guesses, word_instance.word)
|
|
||||||
if game_over:
|
|
||||||
await complete_wordle_game(session, interaction.user.id, word_instance.word in guesses)
|
|
||||||
|
|
||||||
|
|
||||||
async def setup(client: Didier):
|
async def setup(client: Didier):
|
||||||
"""Load the cog"""
|
"""Load the cog"""
|
||||||
|
|
|
@ -10,7 +10,6 @@ from database import enums
|
||||||
from database.crud.birthdays import get_birthdays_on_day
|
from database.crud.birthdays import get_birthdays_on_day
|
||||||
from database.crud.reminders import get_all_reminders_for_category
|
from database.crud.reminders import get_all_reminders_for_category
|
||||||
from database.crud.ufora_announcements import remove_old_announcements
|
from database.crud.ufora_announcements import remove_old_announcements
|
||||||
from database.crud.wordle import set_daily_word
|
|
||||||
from database.schemas import Reminder
|
from database.schemas import Reminder
|
||||||
from didier import Didier
|
from didier import Didier
|
||||||
from didier.data.embeds.schedules import (
|
from didier.data.embeds.schedules import (
|
||||||
|
@ -54,7 +53,6 @@ class Tasks(commands.Cog):
|
||||||
"reminders": self.reminders,
|
"reminders": self.reminders,
|
||||||
"ufora": self.pull_ufora_announcements,
|
"ufora": self.pull_ufora_announcements,
|
||||||
"remove_ufora": self.remove_old_ufora_announcements,
|
"remove_ufora": self.remove_old_ufora_announcements,
|
||||||
"wordle": self.reset_wordle_word,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
@overrides
|
@overrides
|
||||||
|
@ -74,7 +72,6 @@ class Tasks(commands.Cog):
|
||||||
|
|
||||||
# Start other tasks
|
# Start other tasks
|
||||||
self.reminders.start()
|
self.reminders.start()
|
||||||
self.reset_wordle_word.start()
|
|
||||||
|
|
||||||
@overrides
|
@overrides
|
||||||
def cog_unload(self) -> None:
|
def cog_unload(self) -> None:
|
||||||
|
@ -266,34 +263,16 @@ class Tasks(commands.Cog):
|
||||||
async with self.client.postgres_session as session:
|
async with self.client.postgres_session as session:
|
||||||
await remove_old_announcements(session)
|
await remove_old_announcements(session)
|
||||||
|
|
||||||
@tasks.loop(time=DAILY_RESET_TIME)
|
|
||||||
async def reset_wordle_word(self, forced: bool = False):
|
|
||||||
"""Reset the daily Wordle word"""
|
|
||||||
async with self.client.postgres_session as session:
|
|
||||||
await set_daily_word(session, random.choice(tuple(self.client.wordle_words)), forced=forced)
|
|
||||||
await self.client.database_caches.wordle_word.invalidate(session)
|
|
||||||
|
|
||||||
@reset_wordle_word.before_loop
|
|
||||||
async def _before_reset_wordle_word(self):
|
|
||||||
await self.client.wait_until_ready()
|
|
||||||
|
|
||||||
@check_birthdays.error
|
@check_birthdays.error
|
||||||
@pull_schedules.error
|
@pull_schedules.error
|
||||||
@pull_ufora_announcements.error
|
@pull_ufora_announcements.error
|
||||||
@reminders.error
|
@reminders.error
|
||||||
@remove_old_ufora_announcements.error
|
@remove_old_ufora_announcements.error
|
||||||
@reset_wordle_word.error
|
|
||||||
async def _on_tasks_error(self, error: BaseException):
|
async def _on_tasks_error(self, error: BaseException):
|
||||||
"""Error handler for all tasks"""
|
"""Error handler for all tasks"""
|
||||||
self.client.dispatch("task_error", error)
|
self.client.dispatch("task_error", error)
|
||||||
|
|
||||||
|
|
||||||
async def setup(client: Didier):
|
async def setup(client: Didier):
|
||||||
"""Load the cog
|
"""Load the cog"""
|
||||||
|
await client.add_cog(Tasks(client))
|
||||||
Initially fetch the wordle word from the database, or reset it
|
|
||||||
if there hasn't been a reset yet today
|
|
||||||
"""
|
|
||||||
cog = Tasks(client)
|
|
||||||
await client.add_cog(cog)
|
|
||||||
await cog.reset_wordle_word()
|
|
||||||
|
|
|
@ -1,142 +0,0 @@
|
||||||
import enum
|
|
||||||
from dataclasses import dataclass
|
|
||||||
|
|
||||||
import discord
|
|
||||||
from overrides import overrides
|
|
||||||
|
|
||||||
from database.constants import WORDLE_GUESS_COUNT, WORDLE_WORD_LENGTH
|
|
||||||
from database.schemas import WordleWord
|
|
||||||
from didier.data.embeds.base import EmbedBaseModel
|
|
||||||
from didier.utils.types.datetime import int_to_weekday, tz_aware_now
|
|
||||||
|
|
||||||
__all__ = ["is_wordle_game_over", "WordleEmbed", "WordleErrorEmbed"]
|
|
||||||
|
|
||||||
|
|
||||||
def is_wordle_game_over(guesses: list[str], word: str) -> bool:
|
|
||||||
"""Check if the current game is over or not"""
|
|
||||||
if not guesses:
|
|
||||||
return False
|
|
||||||
|
|
||||||
if len(guesses) == WORDLE_GUESS_COUNT:
|
|
||||||
return True
|
|
||||||
|
|
||||||
return word.lower() in guesses
|
|
||||||
|
|
||||||
|
|
||||||
def footer() -> str:
|
|
||||||
"""Create the footer to put on the embed"""
|
|
||||||
today = tz_aware_now()
|
|
||||||
return f"{int_to_weekday(today.weekday())} {today.strftime('%d/%m/%Y')}"
|
|
||||||
|
|
||||||
|
|
||||||
class WordleColour(enum.IntEnum):
|
|
||||||
"""Colours for the Wordle embed"""
|
|
||||||
|
|
||||||
EMPTY = 0
|
|
||||||
WRONG_LETTER = 1
|
|
||||||
WRONG_POSITION = 2
|
|
||||||
CORRECT = 3
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
|
||||||
class WordleEmbed(EmbedBaseModel):
|
|
||||||
"""Embed for a Wordle game"""
|
|
||||||
|
|
||||||
guesses: list[str]
|
|
||||||
word: WordleWord
|
|
||||||
|
|
||||||
def _letter_colour(self, guess: str, index: int) -> WordleColour:
|
|
||||||
"""Get the colour for a guess at a given position"""
|
|
||||||
if guess[index] == self.word.word[index]:
|
|
||||||
return WordleColour.CORRECT
|
|
||||||
|
|
||||||
wrong_letter = 0
|
|
||||||
wrong_position = 0
|
|
||||||
|
|
||||||
for i, letter in enumerate(self.word.word):
|
|
||||||
if letter == guess[index] and guess[i] != guess[index]:
|
|
||||||
wrong_letter += 1
|
|
||||||
|
|
||||||
if i <= index and guess[i] == guess[index] and letter != guess[index]:
|
|
||||||
wrong_position += 1
|
|
||||||
|
|
||||||
if i >= index:
|
|
||||||
if wrong_position == 0:
|
|
||||||
break
|
|
||||||
|
|
||||||
if wrong_position <= wrong_letter:
|
|
||||||
return WordleColour.WRONG_POSITION
|
|
||||||
|
|
||||||
return WordleColour.WRONG_LETTER
|
|
||||||
|
|
||||||
def _guess_colours(self, guess: str) -> list[WordleColour]:
|
|
||||||
"""Create the colour codes for a specific guess"""
|
|
||||||
return [self._letter_colour(guess, i) for i in range(WORDLE_WORD_LENGTH)]
|
|
||||||
|
|
||||||
def colour_code_game(self) -> list[list[WordleColour]]:
|
|
||||||
"""Create the colour codes for an entire game"""
|
|
||||||
colours = []
|
|
||||||
|
|
||||||
# Add all the guesses
|
|
||||||
for guess in self.guesses:
|
|
||||||
colours.append(self._guess_colours(guess))
|
|
||||||
|
|
||||||
# Fill the rest with empty spots
|
|
||||||
for _ in range(WORDLE_GUESS_COUNT - len(colours)):
|
|
||||||
colours.append([WordleColour.EMPTY] * WORDLE_WORD_LENGTH)
|
|
||||||
|
|
||||||
return colours
|
|
||||||
|
|
||||||
def _colours_to_emojis(self, colours: list[list[WordleColour]]) -> list[list[str]]:
|
|
||||||
"""Turn the colours of the board into Discord emojis"""
|
|
||||||
colour_map = {
|
|
||||||
WordleColour.EMPTY: ":white_large_square:",
|
|
||||||
WordleColour.WRONG_LETTER: ":black_large_square:",
|
|
||||||
WordleColour.WRONG_POSITION: ":orange_square:",
|
|
||||||
WordleColour.CORRECT: ":green_square:",
|
|
||||||
}
|
|
||||||
|
|
||||||
emojis = []
|
|
||||||
for row in colours:
|
|
||||||
emojis.append(list(map(lambda char: colour_map[char], row)))
|
|
||||||
|
|
||||||
return emojis
|
|
||||||
|
|
||||||
@overrides
|
|
||||||
def to_embed(self, **kwargs) -> discord.Embed:
|
|
||||||
only_colours = kwargs.get("only_colours", False)
|
|
||||||
|
|
||||||
colours = self.colour_code_game()
|
|
||||||
|
|
||||||
embed = discord.Embed(colour=discord.Colour.blue(), title=f"Wordle #{self.word.word_id + 1}")
|
|
||||||
emojis = self._colours_to_emojis(colours)
|
|
||||||
|
|
||||||
rows = [" ".join(row) for row in emojis]
|
|
||||||
|
|
||||||
# Don't reveal anything if we only want to show the colours
|
|
||||||
if not only_colours and self.guesses:
|
|
||||||
for i, guess in enumerate(self.guesses):
|
|
||||||
rows[i] += f" ||{guess.upper()}||"
|
|
||||||
|
|
||||||
# If the game is over, reveal the word
|
|
||||||
if is_wordle_game_over(self.guesses, self.word.word):
|
|
||||||
rows.append(f"\n\nThe word was **{self.word.word.upper()}**!")
|
|
||||||
|
|
||||||
embed.description = "\n\n".join(rows)
|
|
||||||
embed.set_footer(text=footer())
|
|
||||||
|
|
||||||
return embed
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
|
||||||
class WordleErrorEmbed(EmbedBaseModel):
|
|
||||||
"""Embed to send error messages to the user"""
|
|
||||||
|
|
||||||
message: str
|
|
||||||
|
|
||||||
@overrides
|
|
||||||
def to_embed(self, **kwargs) -> discord.Embed:
|
|
||||||
embed = discord.Embed(colour=discord.Colour.red(), title="Wordle")
|
|
||||||
embed.description = self.message
|
|
||||||
embed.set_footer(text=footer())
|
|
||||||
return embed
|
|
|
@ -38,7 +38,6 @@ class Didier(commands.Bot):
|
||||||
http_session: ClientSession
|
http_session: ClientSession
|
||||||
schedules: dict[settings.ScheduleType, Schedule] = {}
|
schedules: dict[settings.ScheduleType, Schedule] = {}
|
||||||
sniped: dict[int, tuple[discord.Message, Optional[discord.Message]]] = {}
|
sniped: dict[int, tuple[discord.Message, Optional[discord.Message]]] = {}
|
||||||
wordle_words: set[str] = set()
|
|
||||||
|
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
activity = discord.Activity(type=discord.ActivityType.playing, name=settings.DISCORD_STATUS_MESSAGE)
|
activity = discord.Activity(type=discord.ActivityType.playing, name=settings.DISCORD_STATUS_MESSAGE)
|
||||||
|
@ -77,9 +76,6 @@ class Didier(commands.Bot):
|
||||||
# Create directories that are ignored on GitHub
|
# Create directories that are ignored on GitHub
|
||||||
self._create_ignored_directories()
|
self._create_ignored_directories()
|
||||||
|
|
||||||
# Load the Wordle dictionary
|
|
||||||
self._load_wordle_words()
|
|
||||||
|
|
||||||
# Initialize caches
|
# Initialize caches
|
||||||
self.database_caches = CacheManager()
|
self.database_caches = CacheManager()
|
||||||
async with self.postgres_session as session:
|
async with self.postgres_session as session:
|
||||||
|
@ -137,12 +133,6 @@ class Didier(commands.Bot):
|
||||||
elif os.path.isdir(new_path := f"{path}/{file}"):
|
elif os.path.isdir(new_path := f"{path}/{file}"):
|
||||||
await self._load_directory_extensions(new_path)
|
await self._load_directory_extensions(new_path)
|
||||||
|
|
||||||
def _load_wordle_words(self):
|
|
||||||
"""Load the dictionary of Wordle words"""
|
|
||||||
with open("files/dictionaries/words-english-wordle.txt", "r") as fp:
|
|
||||||
for line in fp:
|
|
||||||
self.wordle_words.add(line.strip())
|
|
||||||
|
|
||||||
async def load_schedules(self):
|
async def load_schedules(self):
|
||||||
"""Parse & load all schedules into memory"""
|
"""Parse & load all schedules into memory"""
|
||||||
self.schedules = {}
|
self.schedules = {}
|
||||||
|
|
File diff suppressed because it is too large
Load Diff
|
@ -1,138 +0,0 @@
|
||||||
from datetime import date, timedelta
|
|
||||||
|
|
||||||
import pytest
|
|
||||||
from freezegun import freeze_time
|
|
||||||
from sqlalchemy.ext.asyncio import AsyncSession
|
|
||||||
|
|
||||||
from database.crud import wordle as crud
|
|
||||||
from database.schemas import User, WordleGuess, WordleWord
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
|
||||||
async def wordle_guesses(postgres: AsyncSession, user: User) -> list[WordleGuess]:
|
|
||||||
"""Fixture to generate some guesses"""
|
|
||||||
guesses = []
|
|
||||||
|
|
||||||
for guess in ["TEST", "WORDLE", "WORDS"]:
|
|
||||||
guess = WordleGuess(user_id=user.user_id, guess=guess)
|
|
||||||
postgres.add(guess)
|
|
||||||
await postgres.commit()
|
|
||||||
|
|
||||||
guesses.append(guess)
|
|
||||||
|
|
||||||
return guesses
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.postgres
|
|
||||||
async def test_get_active_wordle_game_none(postgres: AsyncSession, user: User):
|
|
||||||
"""Test getting an active game when there is none"""
|
|
||||||
result = await crud.get_active_wordle_game(postgres, user.user_id)
|
|
||||||
assert not result
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.postgres
|
|
||||||
async def test_get_active_wordle_game(postgres: AsyncSession, wordle_guesses: list[WordleGuess]):
|
|
||||||
"""Test getting an active game when there is one"""
|
|
||||||
result = await crud.get_active_wordle_game(postgres, wordle_guesses[0].user_id)
|
|
||||||
assert result == wordle_guesses
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.postgres
|
|
||||||
async def test_get_daily_word_none(postgres: AsyncSession):
|
|
||||||
"""Test getting the daily word when the database is empty"""
|
|
||||||
result = await crud.get_daily_word(postgres)
|
|
||||||
assert result is None
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.postgres
|
|
||||||
@freeze_time("2022-07-30")
|
|
||||||
async def test_get_daily_word_not_today(postgres: AsyncSession):
|
|
||||||
"""Test getting the daily word when there is an entry, but not for today"""
|
|
||||||
day = date.today() - timedelta(days=1)
|
|
||||||
|
|
||||||
word = "testword"
|
|
||||||
word_instance = WordleWord(word=word, day=day)
|
|
||||||
postgres.add(word_instance)
|
|
||||||
await postgres.commit()
|
|
||||||
|
|
||||||
assert await crud.get_daily_word(postgres) is None
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.postgres
|
|
||||||
@freeze_time("2022-07-30")
|
|
||||||
async def test_get_daily_word_present(postgres: AsyncSession):
|
|
||||||
"""Test getting the daily word when there is one for today"""
|
|
||||||
day = date.today()
|
|
||||||
|
|
||||||
word = "testword"
|
|
||||||
word_instance = WordleWord(word=word, day=day)
|
|
||||||
postgres.add(word_instance)
|
|
||||||
await postgres.commit()
|
|
||||||
|
|
||||||
daily_word = await crud.get_daily_word(postgres)
|
|
||||||
assert daily_word is not None
|
|
||||||
assert daily_word.word == word
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.postgres
|
|
||||||
@freeze_time("2022-07-30")
|
|
||||||
async def test_set_daily_word_none_present(postgres: AsyncSession):
|
|
||||||
"""Test setting the daily word when there is none"""
|
|
||||||
assert await crud.get_daily_word(postgres) is None
|
|
||||||
word = "testword"
|
|
||||||
await crud.set_daily_word(postgres, word)
|
|
||||||
|
|
||||||
daily_word = await crud.get_daily_word(postgres)
|
|
||||||
assert daily_word is not None
|
|
||||||
assert daily_word.word == word
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.postgres
|
|
||||||
@freeze_time("2022-07-30")
|
|
||||||
async def test_set_daily_word_present(postgres: AsyncSession):
|
|
||||||
"""Test setting the daily word when there already is one"""
|
|
||||||
word = "testword"
|
|
||||||
await crud.set_daily_word(postgres, word)
|
|
||||||
await crud.set_daily_word(postgres, "another word")
|
|
||||||
|
|
||||||
daily_word = await crud.get_daily_word(postgres)
|
|
||||||
assert daily_word is not None
|
|
||||||
assert daily_word.word == word
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.postgres
|
|
||||||
@freeze_time("2022-07-30")
|
|
||||||
async def test_set_daily_word_force_overwrite(postgres: AsyncSession):
|
|
||||||
"""Test setting the daily word when there already is one, but "forced" is set to True"""
|
|
||||||
word = "testword"
|
|
||||||
await crud.set_daily_word(postgres, word)
|
|
||||||
word = "anotherword"
|
|
||||||
await crud.set_daily_word(postgres, word, forced=True)
|
|
||||||
|
|
||||||
daily_word = await crud.get_daily_word(postgres)
|
|
||||||
assert daily_word is not None
|
|
||||||
assert daily_word.word == word
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.postgres
|
|
||||||
async def test_make_wordle_guess(postgres: AsyncSession, user: User):
|
|
||||||
"""Test making a guess in your current game"""
|
|
||||||
test_user_id = user.user_id
|
|
||||||
|
|
||||||
guess = "guess"
|
|
||||||
await crud.make_wordle_guess(postgres, test_user_id, guess)
|
|
||||||
assert await crud.get_wordle_guesses(postgres, test_user_id) == [guess]
|
|
||||||
|
|
||||||
other_guess = "otherguess"
|
|
||||||
await crud.make_wordle_guess(postgres, test_user_id, other_guess)
|
|
||||||
assert await crud.get_wordle_guesses(postgres, test_user_id) == [guess, other_guess]
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.postgres
|
|
||||||
async def test_reset_wordle_games(postgres: AsyncSession, wordle_guesses: list[WordleGuess], user: User):
|
|
||||||
"""Test dropping the collection of active games"""
|
|
||||||
test_user_id = user.user_id
|
|
||||||
|
|
||||||
assert await crud.get_active_wordle_game(postgres, test_user_id)
|
|
||||||
await crud.reset_wordle_games(postgres)
|
|
||||||
assert not await crud.get_active_wordle_game(postgres, test_user_id)
|
|
|
@ -1,72 +0,0 @@
|
||||||
import datetime
|
|
||||||
|
|
||||||
import pytest
|
|
||||||
from freezegun import freeze_time
|
|
||||||
from sqlalchemy import select
|
|
||||||
from sqlalchemy.ext.asyncio import AsyncSession
|
|
||||||
|
|
||||||
from database.crud import wordle_stats as crud
|
|
||||||
from database.schemas import User, WordleStats
|
|
||||||
|
|
||||||
|
|
||||||
async def insert_game_stats(postgres: AsyncSession, stats: WordleStats):
|
|
||||||
"""Helper function to insert some stats"""
|
|
||||||
postgres.add(stats)
|
|
||||||
await postgres.commit()
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.postgres
|
|
||||||
async def test_get_stats_non_existent_creates(postgres: AsyncSession, user: User):
|
|
||||||
"""Test getting a user's stats when the db is empty"""
|
|
||||||
test_user_id = user.user_id
|
|
||||||
|
|
||||||
statement = select(WordleStats).where(WordleStats.user_id == test_user_id)
|
|
||||||
assert (await postgres.execute(statement)).scalar_one_or_none() is None
|
|
||||||
|
|
||||||
await crud.get_wordle_stats(postgres, test_user_id)
|
|
||||||
assert (await postgres.execute(statement)).scalar_one_or_none() is not None
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.postgres
|
|
||||||
async def test_get_stats_existing_returns(postgres: AsyncSession, user: User):
|
|
||||||
"""Test getting a user's stats when there's already an entry present"""
|
|
||||||
test_user_id = user.user_id
|
|
||||||
|
|
||||||
stats = WordleStats(user_id=test_user_id)
|
|
||||||
stats.games = 20
|
|
||||||
await insert_game_stats(postgres, stats)
|
|
||||||
found_stats = await crud.get_wordle_stats(postgres, test_user_id)
|
|
||||||
assert found_stats.games == 20
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.postgres
|
|
||||||
@freeze_time("2022-07-30")
|
|
||||||
async def test_complete_wordle_game_won(postgres: AsyncSession, user: User):
|
|
||||||
"""Test completing a wordle game when you win"""
|
|
||||||
test_user_id = user.user_id
|
|
||||||
|
|
||||||
await crud.complete_wordle_game(postgres, test_user_id, win=True)
|
|
||||||
stats = await crud.get_wordle_stats(postgres, test_user_id)
|
|
||||||
assert stats.games == 1
|
|
||||||
assert stats.wins == 1
|
|
||||||
assert stats.current_streak == 1
|
|
||||||
assert stats.highest_streak == 1
|
|
||||||
assert stats.last_win == datetime.date.today()
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.postgres
|
|
||||||
@freeze_time("2022-07-30")
|
|
||||||
async def test_complete_wordle_game_lost(postgres: AsyncSession, user: User):
|
|
||||||
"""Test completing a wordle game when you lose"""
|
|
||||||
test_user_id = user.user_id
|
|
||||||
|
|
||||||
stats = WordleStats(user_id=test_user_id)
|
|
||||||
stats.current_streak = 10
|
|
||||||
await insert_game_stats(postgres, stats)
|
|
||||||
|
|
||||||
await crud.complete_wordle_game(postgres, test_user_id, win=False)
|
|
||||||
stats = await crud.get_wordle_stats(postgres, test_user_id)
|
|
||||||
|
|
||||||
# Check that streak was broken
|
|
||||||
assert stats.current_streak == 0
|
|
||||||
assert stats.games == 1
|
|
Loading…
Reference in New Issue