| 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
 | from contextlib import suppress
from functools import partial
from pathlib import Path
import discord
import yaml
from discord.ext import commands
from bot.bot import Bot
from bot.constants import MODERATION_ROLES, WHITELISTED_CHANNELS
from bot.utils.decorators import whitelist_override
from bot.utils.randomization import RandomCycle
SUGGESTION_FORM = "https://forms.gle/zw6kkJqv8U43Nfjg9"
with Path("bot/resources/utilities/starter.yaml").open("r", encoding="utf8") as f:
    STARTERS = yaml.safe_load(f)
with Path("bot/resources/utilities/py_topics.yaml").open("r", encoding="utf8") as f:
    # First ID is #python-general and the rest are top to bottom categories of Topical Chat/Help.
    PY_TOPICS = yaml.safe_load(f)
    # Removing `None` from lists of topics, if not a list, it is changed to an empty one.
    PY_TOPICS = {k: [i for i in v if i] if isinstance(v, list) else [] for k, v in PY_TOPICS.items()}
    # All the allowed channels that the ".topic" command is allowed to be executed in.
    ALL_ALLOWED_CHANNELS = list(PY_TOPICS.keys()) + list(WHITELISTED_CHANNELS)
# Putting all topics into one dictionary and shuffling lists to reduce same-topic repetitions.
ALL_TOPICS = {"default": STARTERS, **PY_TOPICS}
TOPICS = {
    channel: RandomCycle(topics or ["No topics found for this channel."])
    for channel, topics in ALL_TOPICS.items()
}
class ConvoStarters(commands.Cog):
    """General conversation topics."""
    def __init__(self, bot: Bot):
        self.bot = bot
    @staticmethod
    def _build_topic_embed(channel_id: int) -> discord.Embed:
        """
        Build an embed containing a conversation topic.
        If in a Python channel, a python-related topic will be given.
        Otherwise, a random conversation topic will be received by the user.
        """
        # No matter what, the form will be shown.
        embed = discord.Embed(
            description=f"Suggest more topics [here]({SUGGESTION_FORM})!",
            color=discord.Colour.og_blurple()
        )
        try:
            channel_topics = TOPICS[channel_id]
        except KeyError:
            # Channel doesn't have any topics.
            embed.title = f"**{next(TOPICS['default'])}**"
        else:
            embed.title = f"**{next(channel_topics)}**"
        return embed
    @staticmethod
    def _predicate(
        command_invoker: discord.User | discord.Member,
        message: discord.Message,
        reaction: discord.Reaction,
        user: discord.User
    ) -> bool:
        user_is_moderator = any(role.id in MODERATION_ROLES for role in getattr(user, "roles", []))
        user_is_invoker = user.id == command_invoker.id
        is_right_reaction = all((
            reaction.message.id == message.id,
            str(reaction.emoji) == "🔄",
            user_is_moderator or user_is_invoker
        ))
        return is_right_reaction
    async def _listen_for_refresh(
        self,
        command_invoker: discord.User | discord.Member,
        message: discord.Message
    ) -> None:
        await message.add_reaction("🔄")
        while True:
            try:
                reaction, user = await self.bot.wait_for(
                    "reaction_add",
                    check=partial(self._predicate, command_invoker, message),
                    timeout=60.0
                )
            except TimeoutError:
                with suppress(discord.NotFound):
                    await message.clear_reaction("🔄")
                break
            try:
                await message.edit(embed=self._build_topic_embed(message.channel.id))
            except discord.NotFound:
                break
            with suppress(discord.NotFound):
                await message.remove_reaction(reaction, user)
    @commands.command()
    @commands.cooldown(1, 60*2, commands.BucketType.channel)
    @whitelist_override(channels=ALL_ALLOWED_CHANNELS)
    async def topic(self, ctx: commands.Context) -> None:
        """
        Responds with a random topic to start a conversation.
        Allows the refresh of a topic by pressing an emoji.
        """
        message = await ctx.send(embed=self._build_topic_embed(ctx.channel.id))
        self.bot.loop.create_task(self._listen_for_refresh(ctx.author, message))
async def setup(bot: Bot) -> None:
    """Load the ConvoStarters cog."""
    await bot.add_cog(ConvoStarters(bot))
 |