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()