mirror of https://github.com/stijndcl/didier
86 lines
2.6 KiB
Python
86 lines
2.6 KiB
Python
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()
|