From 355e036f6d0ed4d29d91faac12975f8359e4cf35 Mon Sep 17 00:00:00 2001 From: pushiscool Date: Mon, 25 May 2026 22:11:50 -0400 Subject: [PATCH] Trust Wordle bot summaries for leaderboard --- src/cogs/wordle.py | 93 +++++++++++++++++---------------- src/database_handler.py | 15 ++++-- tests/test_database_handler.py | 9 ++-- tests/test_wordle_helpers.py | 94 ++++++++++++++++++++++++++++++++++ 4 files changed, 157 insertions(+), 54 deletions(-) diff --git a/src/cogs/wordle.py b/src/cogs/wordle.py index 1161475..1f89a74 100644 --- a/src/cogs/wordle.py +++ b/src/cogs/wordle.py @@ -13,6 +13,7 @@ COMMAND_GUILD_IDS = get_command_guild_ids(conf) WORDLE_RESULT_RE = re.compile(r"\bWordle\s+(\d[\d,]*)\s+([1-6X])/6(\*)?", re.IGNORECASE) +WORDLE_PUZZLE_RE = re.compile(r"\bWordle\s+(\d[\d,]*)\b", re.IGNORECASE) WORDLE_SUMMARY_RE = re.compile(r"\b([1-6X])/6(\*)?(?=\s|[:\-–—]|$)\s*(?::|[-–—])?\s*((?:<@!?\d+>[\s,;]*)+)", re.IGNORECASE) MENTION_RE = re.compile(r"<@!?(\d+)>") @@ -55,6 +56,33 @@ def is_wordle_channel(channel) -> bool: return (getattr(channel, "name", "") or "").lower() == "wordle" +def configured_wordle_bot_id(): + value = conf.get("wordle_bot_id") + if value is None: + return None + + try: + return int(value) + except (TypeError, ValueError): + return None + + +def is_wordle_bot_author(author) -> bool: + if not getattr(author, "bot", False): + return False + + expected_id = configured_wordle_bot_id() + if expected_id is not None: + return getattr(author, "id", None) == expected_id + + names = ( + getattr(author, "name", ""), + getattr(author, "display_name", ""), + getattr(author, "global_name", ""), + ) + return any("wordle" in (name or "").lower() for name in names) + + def get_history_bounds(start, end): after = datetime(start.year, start.month, start.day, tzinfo=timezone.utc) - timedelta(seconds=1) end_plus_two = end + timedelta(days=2) @@ -108,6 +136,14 @@ def parse_wordle_result(content: str): } +def parse_wordle_puzzle(content: str): + match = WORDLE_PUZZLE_RE.search(content) + if not match: + return None + + return int(match.group(1).replace(",", "")) + + def parse_wordle_summary_text(content: str): entries = [] @@ -234,8 +270,6 @@ async def sync_channel_history(self, channel, start, end) -> int: processed = 0 async for message in channel.history(limit=None, after=after, before=before, oldest_first=True): - if await self.process_wordle_result_message(message): - processed += 1 processed += await self.process_wordle_summary_message(message) return processed @@ -286,50 +320,21 @@ async def wordle_leaderboard(self, inter: Interaction) -> None: await inter.send(embed=embed) async def process_wordle_result_message(self, message: nextcord.Message) -> bool: - if message.guild is None or message.author.bot: - return False - - if not is_wordle_channel(message.channel): - return False - - result = parse_wordle_result(message.content) - if result is None: - return False - - season = await self.bot.db.wordle.get_active_season(message.guild.id) - if not season: - return False - - played_date = get_message_date(message).isoformat() - if not date_in_season(played_date, season): - return False - - await self.bot.db.wordle.save_result( - guild_id=message.guild.id, - channel_id=message.channel.id, - user_id=message.author.id, - username=message.author.display_name, - puzzle=result["puzzle"], - tries=result["tries"], - failed=result["failed"], - hard_mode=result["hard_mode"], - score=result["score"], - played_date=played_date, - message_id=message.id, - season_start=season["start_date"], - season_end=season["end_date"], - source="user_share", - ) - return True + # Only the Wordle bot's daily summary should update the leaderboard. + # Member share messages are easy to spoof and can use the wrong puzzle. + return False async def process_wordle_summary_message(self, message: nextcord.Message) -> int: - if message.guild is None or not message.author.bot: + if message.guild is None or not is_wordle_bot_author(message.author): return 0 if not is_wordle_channel(message.channel): return 0 text = get_message_text(message) + if not is_wordle_summary_message(text): + return 0 + entries = parse_wordle_summary_text(text) if not entries: @@ -345,6 +350,7 @@ async def process_wordle_summary_message(self, message: nextcord.Message) -> int return 0 count = 0 + puzzle = parse_wordle_puzzle(text) for entry in entries: username = await self.resolve_username(message.guild, entry["user_id"]) @@ -353,7 +359,7 @@ async def process_wordle_summary_message(self, message: nextcord.Message) -> int channel_id=message.channel.id, user_id=entry["user_id"], username=username, - puzzle=None, + puzzle=puzzle, tries=entry["tries"], failed=entry["failed"], hard_mode=entry["hard_mode"], @@ -382,11 +388,6 @@ async def post_leaderboard(self, message: nextcord.Message) -> None: return processed = await self.process_wordle_summary_message(message) - async for recent_message in message.channel.history(limit=100): - if get_message_date(recent_message).isoformat() != played_date: - continue - if await self.process_wordle_result_message(recent_message): - processed += 1 if processed == 0: return @@ -408,9 +409,7 @@ async def post_leaderboard(self, message: nextcord.Message) -> None: @commands.Cog.listener() async def on_message(self, message: nextcord.Message) -> None: - await self.process_wordle_result_message(message) - - if message.guild and message.author.bot and is_wordle_channel(message.channel): + if message.guild and is_wordle_bot_author(message.author) and is_wordle_channel(message.channel): if is_wordle_summary_message(get_message_text(message)): await asyncio.sleep(5) await self.post_leaderboard(message) diff --git a/src/database_handler.py b/src/database_handler.py index c313476..d1203f2 100644 --- a/src/database_handler.py +++ b/src/database_handler.py @@ -783,16 +783,25 @@ async def save_result( season_end: str, source: str, ) -> None: - query = { + base_query = { "guild_id": guild_id, "user_id": user_id, - "played_date": played_date, "season_start": season_start, "season_end": season_end, } + query = dict(base_query) + if puzzle is not None: + query["puzzle"] = puzzle + else: + query["played_date"] = played_date existing = await self.wordle_results.find_one(query) - if existing and existing.get("source") == "user_share" and source == "wordle_bot_summary" and not hard_mode: + if not existing and puzzle is not None: + query = dict(base_query) + query["played_date"] = played_date + existing = await self.wordle_results.find_one(query) + + if existing and existing.get("source") == "wordle_bot_summary" and source != "wordle_bot_summary": return await self.wordle_results.update_one( diff --git a/tests/test_database_handler.py b/tests/test_database_handler.py index d02d52a..d055d38 100644 --- a/tests/test_database_handler.py +++ b/tests/test_database_handler.py @@ -242,13 +242,14 @@ def test_wordle_leaderboard_sums_scores_and_sorts_lowest_first(): assert rows[1]["failures"] == 1 -def test_wordle_summary_does_not_downgrade_hard_user_share(): +def test_wordle_summary_replaces_user_share_for_same_puzzle(): db = make_wordle_db() asyncio.run(db.save_result(1, 10, 111, "Push", 1801, 3, False, True, 2, "2026-05-01", 100, "2026-05-01", "2026-05-31", "user_share")) - asyncio.run(db.save_result(1, 10, 111, "Push", None, 3, False, False, 3, "2026-05-01", 101, "2026-05-01", "2026-05-31", "wordle_bot_summary")) + asyncio.run(db.save_result(1, 10, 111, "Push", 1801, 4, False, False, 4, "2026-05-01", 101, "2026-05-01", "2026-05-31", "wordle_bot_summary")) rows = asyncio.run(db.get_leaderboard(1, "2026-05-01", "2026-05-31")) - assert rows[0]["total_score"] == 2 - assert rows[0]["hard_games"] == 1 + assert rows[0]["total_score"] == 4 + assert rows[0]["hard_games"] == 0 + assert db.wordle_results.docs[0]["source"] == "wordle_bot_summary" diff --git a/tests/test_wordle_helpers.py b/tests/test_wordle_helpers.py index 4ebe5f6..dc599d1 100644 --- a/tests/test_wordle_helpers.py +++ b/tests/test_wordle_helpers.py @@ -1,12 +1,16 @@ import asyncio +from datetime import datetime, timezone from types import SimpleNamespace +from unittest.mock import AsyncMock from cogs.wordle import ( Wordle, date_in_season, format_wordle_leaderboard, + is_wordle_bot_author, is_wordle_summary_message, parse_date, + parse_wordle_puzzle, parse_wordle_result, parse_wordle_summary_text, wordle_score, @@ -43,6 +47,11 @@ def test_parse_wordle_result_failed(): assert result["score"] == 7 +def test_parse_wordle_puzzle(): + assert parse_wordle_puzzle("Wordle 1,800\nHere are yesterday's results") == 1800 + assert parse_wordle_puzzle("Here are yesterday's results") is None + + def test_parse_wordle_result_invalid(): assert parse_wordle_result("push was playing") is None @@ -73,6 +82,12 @@ def test_is_wordle_summary_message(): assert is_wordle_summary_message("hello") is False +def test_is_wordle_bot_author_uses_bot_name(): + assert is_wordle_bot_author(SimpleNamespace(bot=True, name="Wordle")) is True + assert is_wordle_bot_author(SimpleNamespace(bot=True, name="APBot")) is False + assert is_wordle_bot_author(SimpleNamespace(bot=False, name="Wordle")) is False + + def test_wordle_score(): assert wordle_score(3, False, False) == 3 assert wordle_score(3, False, True) == 2 @@ -117,3 +132,82 @@ async def fetch_member(self, user_id): cog = Wordle(bot=object()) assert asyncio.run(cog.resolve_username(FakeGuild(), 123)) == "Fetched Name" + + +def test_process_wordle_result_message_ignores_member_share(): + wordle_db = SimpleNamespace( + get_active_season=AsyncMock(return_value={"start_date": "2026-05-25", "end_date": "2026-05-25"}), + save_result=AsyncMock(), + ) + bot = SimpleNamespace(db=SimpleNamespace(wordle=wordle_db)) + cog = Wordle(bot=bot) + message = SimpleNamespace( + guild=SimpleNamespace(id=1), + author=SimpleNamespace(bot=False, id=111, display_name="Player"), + channel=SimpleNamespace(id=10, name="wordle"), + content="Wordle 1,800 3/6*", + created_at=datetime(2026, 5, 25, tzinfo=timezone.utc), + id=55, + ) + + processed = asyncio.run(cog.process_wordle_result_message(message)) + + assert processed is False + wordle_db.save_result.assert_not_awaited() + + +def test_process_wordle_summary_requires_wordle_bot(): + wordle_db = SimpleNamespace( + get_active_season=AsyncMock(return_value={"start_date": "2026-05-25", "end_date": "2026-05-25"}), + save_result=AsyncMock(), + ) + bot = SimpleNamespace(db=SimpleNamespace(wordle=wordle_db)) + cog = Wordle(bot=bot) + message = SimpleNamespace( + guild=SimpleNamespace(id=1), + author=SimpleNamespace(bot=True, name="APBot", display_name="APBot"), + channel=SimpleNamespace(id=10, name="wordle"), + content="Wordle 1,800\nHere are yesterday's results:\n3/6*: <@111>", + embeds=[], + created_at=datetime(2026, 5, 26, tzinfo=timezone.utc), + id=56, + ) + + processed = asyncio.run(cog.process_wordle_summary_message(message)) + + assert processed == 0 + wordle_db.save_result.assert_not_awaited() + + +def test_process_wordle_summary_records_puzzle_from_wordle_bot(): + guild = SimpleNamespace( + id=1, + get_member=lambda user_id: SimpleNamespace(display_name="Player"), + ) + wordle_db = SimpleNamespace( + get_active_season=AsyncMock(return_value={"start_date": "2026-05-25", "end_date": "2026-05-25"}), + save_result=AsyncMock(), + ) + bot = SimpleNamespace(db=SimpleNamespace(wordle=wordle_db)) + cog = Wordle(bot=bot) + message = SimpleNamespace( + guild=guild, + author=SimpleNamespace(bot=True, name="Wordle", display_name="Wordle"), + channel=SimpleNamespace(id=10, name="wordle"), + content="Wordle 1,800\nHere are yesterday's results:\n3/6*: <@111>", + embeds=[], + created_at=datetime(2026, 5, 26, tzinfo=timezone.utc), + id=57, + ) + + processed = asyncio.run(cog.process_wordle_summary_message(message)) + + assert processed == 1 + kwargs = wordle_db.save_result.await_args.kwargs + assert kwargs["user_id"] == 111 + assert kwargs["username"] == "Player" + assert kwargs["puzzle"] == 1800 + assert kwargs["hard_mode"] is True + assert kwargs["score"] == 2 + assert kwargs["played_date"] == "2026-05-25" + assert kwargs["source"] == "wordle_bot_summary"