Added polls module.

owlbot/builtin_modules/polls/commands.py

@@ -0,0 +1,284 @@
+# Copyright 2026 Logan Fick
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Chat commands for the polls module."""
+
+from __future__ import annotations
+
+from markupsafe import escape
+
+from owlbot.api import (
+    ChatEvent,
+    CommandContext,
+    EventContext,
+    EventType,
+    on_command,
+    on_event,
+)
+
+from .manager import PollError, get_manager
+
+
+@on_command("poll", requires_moderator=True, cooldown=3)
+async def poll_command(ctx: CommandContext) -> None:
+    """Start the poll creation flow.
+
+    Generates a moderator token and sends the poll creator a private link
+    to the creation form.
+
+    :param ctx: The command context.
+    """
+    manager = get_manager(ctx.module)
+
+    try:
+        mod_token = await manager.begin_creation(ctx.user.id)
+    except PollError as e:
+        await ctx.owncast_client.send_system_message_to_client(
+            ctx.chat_event.client_id, str(e)
+        )
+        return
+
+    url = ctx.routes.url_for(f"/create/{mod_token}")
+    await ctx.owncast_client.send_system_message_to_client(
+        ctx.chat_event.client_id,
+        f'<a href="{url}"><u>Click here to set up a poll</u></a>.',
+        unsanitized=True,
+    )
+
+
+@on_command("endpoll", requires_moderator=True)
+async def end_poll_command(ctx: CommandContext) -> None:
+    """End the active poll and announce results.
+
+    :param ctx: The command context.
+    """
+    manager = get_manager(ctx.module)
+
+    try:
+        await manager.end()
+    except PollError as e:
+        await ctx.owncast_client.send_system_message_to_client(
+            ctx.chat_event.client_id, str(e)
+        )
+
+
+@on_command("cancelpoll", requires_moderator=True)
+async def cancel_poll_command(ctx: CommandContext) -> None:
+    """Cancel the active poll without announcing results.
+
+    :param ctx: The command context.
+    """
+    manager = get_manager(ctx.module)
+
+    try:
+        await manager.cancel()
+    except PollError as e:
+        await ctx.owncast_client.send_system_message_to_client(
+            ctx.chat_event.client_id, str(e)
+        )
+
+
+@on_command("vote")
+async def vote_command(ctx: CommandContext) -> None:
+    """Cast a vote or get a voting link.
+
+    Hides the vote message, then either processes a quick numeric vote
+    (single-select only) or sends a private voting page link.
+
+    :param ctx: The command context.
+    """
+    manager = get_manager(ctx.module)
+
+    # Hide the vote message to keep votes private and prevent chat spam.
+    await ctx.owncast_client.set_message_visibility(
+        [ctx.chat_event.message_id], visible=False
+    )
+
+    poll = manager.active_poll
+
+    if poll is None:
+        await ctx.owncast_client.send_system_message_to_client(
+            ctx.chat_event.client_id, "There is no active poll."
+        )
+        return
+
+    user_id = ctx.user.id
+    token = manager.get_voter_token(
+        user_id,
+        is_moderator=ctx.user.is_moderator,
+        is_authenticated=ctx.user.is_authenticated,
+    )
+    vote_url = ctx.routes.url_for(f"/vote/{token.value}")
+
+    # Check authentication requirement.
+    if poll.requires_auth and not ctx.user.is_authenticated:
+        await ctx.owncast_client.send_system_message_to_client(
+            ctx.chat_event.client_id,
+            "This poll requires an authenticated account to vote. "
+            f'<a href="{vote_url}">'
+            "<u>Click here to view live results</u></a>.",
+            unsanitized=True,
+        )
+        return
+
+    # Check if the user has already voted.
+    if user_id in poll.votes:
+        await ctx.owncast_client.send_system_message_to_client(
+            ctx.chat_event.client_id,
+            "You have already voted in this poll. "
+            f'<a href="{vote_url}"><u>Click here to view live results</u></a>.',
+            unsanitized=True,
+        )
+        return
+
+    args = ctx.args_list
+
+    if args and not poll.allow_multiple:
+        # Quick vote by number (single-select only).
+        try:
+            choice = int(args[0])
+        except ValueError:
+            choice = None
+
+        if choice is None or choice < 1 or choice > len(poll.options):
+            await ctx.owncast_client.send_system_message_to_client(
+                ctx.chat_event.client_id,
+                f"Invalid option number."
+                f" Use !vote 1-{len(poll.options)} to vote, or "
+                f'<a href="{vote_url}">'
+                "<u>click here to cast your vote</u></a>.",
+                unsanitized=True,
+            )
+            return
+
+        # Convert 1-based to 0-based index.
+        selection = choice - 1
+
+        await manager.record_vote(user_id, {selection})
+        await ctx.owncast_client.send_system_message_to_client(
+            ctx.chat_event.client_id,
+            f"Your vote for #{choice} ({escape(poll.options[selection])})"
+            " has been recorded. "
+            f'<a href="{vote_url}"><u>Click here to view live results</u></a>.',
+            unsanitized=True,
+        )
+        return
+
+    # Send voting page link.
+    if args and poll.allow_multiple:
+        message = (
+            "This poll allows multiple selections,"
+            " so votes must be cast through the voting page. "
+            f'<a href="{vote_url}">'
+            "<u>Click here to cast your vote</u></a>."
+        )
+    else:
+        message = f'<a href="{vote_url}"><u>Click here to cast your vote</u></a>.'
+    await ctx.owncast_client.send_system_message_to_client(
+        ctx.chat_event.client_id,
+        message,
+        unsanitized=True,
+    )
+
+
+@on_event(EventType.CHAT)
+async def handle_bare_vote(ctx: EventContext[ChatEvent]) -> None:
+    """Allow voting by typing a bare number in chat during active polls.
+
+    For single-select polls, records the vote directly. For multi-select
+    polls, sends the user a voting page link instead. The message is hidden
+    to keep votes private and prevent chat spam.
+
+    :param ctx: The event context.
+    """
+    manager = get_manager(ctx.module)
+    poll = manager.active_poll
+    if poll is None:
+        return
+
+    event = ctx.event
+
+    # Keep token privilege flags in sync with the user's current status.
+    manager.refresh_user_status(
+        event.user.id,
+        is_moderator=event.user.is_moderator,
+        is_authenticated=event.user.is_authenticated,
+    )
+
+    text = event.raw_body.strip()
+
+    try:
+        choice = int(text)
+    except ValueError:
+        return
+
+    # Validate the number is within the option range.
+    if choice < 1 or choice > len(poll.options):
+        return
+
+    # Hide the message to keep votes private and prevent chat spam.
+    await ctx.owncast_client.set_message_visibility([event.message_id], visible=False)
+    ctx.stop_propagation("bare number vote in active poll")
+
+    token = manager.get_voter_token(
+        event.user.id,
+        is_moderator=event.user.is_moderator,
+        is_authenticated=event.user.is_authenticated,
+    )
+    vote_url = ctx.routes.url_for(f"/vote/{token.value}")
+
+    # Check authentication requirement.
+    if poll.requires_auth and not event.user.is_authenticated:
+        await ctx.owncast_client.send_system_message_to_client(
+            event.client_id,
+            "This poll requires an authenticated account to vote. "
+            f'<a href="{vote_url}">'
+            "<u>Click here to view live results</u></a>.",
+            unsanitized=True,
+        )
+        return
+
+    # Check if the user has already voted.
+    if event.user.id in poll.votes:
+        await ctx.owncast_client.send_system_message_to_client(
+            event.client_id,
+            "You have already voted in this poll. "
+            f'<a href="{vote_url}"><u>Click here to view live results</u></a>.',
+            unsanitized=True,
+        )
+        return
+
+    # Multi-select polls can't use bare number voting; send a link instead.
+    if poll.allow_multiple:
+        await ctx.owncast_client.send_system_message_to_client(
+            event.client_id,
+            "This poll allows multiple selections,"
+            " so votes must be cast through the voting page. "
+            f'<a href="{vote_url}">'
+            "<u>Click here to cast your vote</u></a>.",
+            unsanitized=True,
+        )
+        return
+
+    # Record the vote.
+    selection = choice - 1
+
+    await manager.record_vote(event.user.id, {selection})
+    await ctx.owncast_client.send_system_message_to_client(
+        event.client_id,
+        f"Your vote for #{choice} ({escape(poll.options[selection])})"
+        " has been recorded. "
+        f'<a href="{vote_url}"><u>Click here to view live results</u></a>.',
+        unsanitized=True,
+    )