import asyncio
from typing import Union
import discord
from redbot.core import Config, checks, commands
from redbot.core.bot import Red
from redbot.core.commands import Cog
# Cog: Any = getattr(commands, "Cog", object)
# listener = getattr(commands.Cog, "listener", None) # Trusty + Sinbad
# if listener is None:
# def listener(name=None):
# return lambda x: x
RATE_LIMIT_DELAY = 60 * 10 # If you're willing to risk rate limiting, you can decrease the delay
class InfoChannel(Cog):
"""
Create a channel with updating server info
Less important information about the cog
"""
def __init__(self, bot: Red):
super().__init__()
self.bot = bot
self.config = Config.get_conf(
self, identifier=731101021116710497110110101108, force_registration=True
)
default_guild = {
"channel_id": None,
"botchannel_id": None,
"onlinechannel_id": None,
"member_count": True,
"bot_count": False,
"online_count": False,
}
self.config.register_guild(**default_guild)
self._critical_section_wooah_ = 0
async def red_delete_data_for_user(self, **kwargs):
"""Nothing to delete"""
return
@commands.command()
@checks.admin()
async def infochannel(self, ctx: commands.Context):
"""
Toggle info channel for this server
"""
def check(m):
return (
m.content.upper() in ["Y", "YES", "N", "NO"]
and m.channel == ctx.channel
and m.author == ctx.author
)
guild: discord.Guild = ctx.guild
channel_id = await self.config.guild(guild).channel_id()
channel = None
if channel_id is not None:
channel: Union[discord.VoiceChannel, None] = guild.get_channel(channel_id)
if channel_id is not None and channel is None:
await ctx.send("Info channel has been deleted, recreate it?")
elif channel_id is None:
await ctx.send("Enable info channel on this server?")
else:
await ctx.send("Do you wish to delete current info channels?")
msg = await self.bot.wait_for("message", check=check)
if msg.content.upper() in ["N", "NO"]:
await ctx.send("Cancelled")
return
if channel is None:
try:
await self.make_infochannel(guild)
except discord.Forbidden:
await ctx.send("Failure: Missing permission to create voice channel")
return
else:
await self.delete_all_infochannels(guild)
if not await ctx.tick():
await ctx.send("Done!")
@commands.group()
@checks.admin()
async def infochannelset(self, ctx: commands.Context):
"""
Toggle different types of infochannels
"""
if not ctx.invoked_subcommand:
pass
@infochannelset.command(name="botcount")
async def _infochannelset_botcount(self, ctx: commands.Context, enabled: bool = None):
"""
Toggle an infochannel that shows the amount of bots in the server
"""
guild = ctx.guild
if enabled is None:
enabled = not await self.config.guild(guild).bot_count()
await self.config.guild(guild).bot_count.set(enabled)
await self.make_infochannel(ctx.guild)
if enabled:
await ctx.send("InfoChannel for bot count has been enabled.")
else:
await ctx.send("InfoChannel for bot count has been disabled.")
@infochannelset.command(name="onlinecount")
async def _infochannelset_onlinecount(self, ctx: commands.Context, enabled: bool = None):
"""
Toggle an infochannel that shows the amount of online users in the server
"""
guild = ctx.guild
if enabled is None:
enabled = not await self.config.guild(guild).online_count()
await self.config.guild(guild).online_count.set(enabled)
await self.make_infochannel(ctx.guild)
if enabled:
await ctx.send("InfoChannel for online user count has been enabled.")
else:
await ctx.send("InfoChannel for online user count has been disabled.")
async def make_infochannel(self, guild: discord.Guild):
botcount = await self.config.guild(guild).bot_count()
onlinecount = await self.config.guild(guild).online_count()
overwrites = {
guild.default_role: discord.PermissionOverwrite(connect=False),
guild.me: discord.PermissionOverwrite(manage_channels=True, connect=True),
}
# Remove the old info channel first
channel_id = await self.config.guild(guild).channel_id()
if channel_id is not None:
channel: discord.VoiceChannel = guild.get_channel(channel_id)
if channel:
await channel.delete(reason="InfoChannel delete")
# Then create the new one
channel = await guild.create_voice_channel(
"Total Humans:", reason="InfoChannel make", overwrites=overwrites
)
await self.config.guild(guild).channel_id.set(channel.id)
if botcount:
# Remove the old bot channel first
botchannel_id = await self.config.guild(guild).botchannel_id()
if channel_id is not None:
botchannel: discord.VoiceChannel = guild.get_channel(botchannel_id)
if botchannel:
await botchannel.delete(reason="InfoChannel delete")
# Then create the new one
botchannel = await guild.create_voice_channel(
"Bots:", reason="InfoChannel botcount", overwrites=overwrites
)
await self.config.guild(guild).botchannel_id.set(botchannel.id)
if onlinecount:
# Remove the old online channel first
onlinechannel_id = await self.config.guild(guild).onlinechannel_id()
if channel_id is not None:
onlinechannel: discord.VoiceChannel = guild.get_channel(onlinechannel_id)
if onlinechannel:
await onlinechannel.delete(reason="InfoChannel delete")
# Then create the new one
onlinechannel = await guild.create_voice_channel(
"Online:", reason="InfoChannel onlinecount", overwrites=overwrites
)
await self.config.guild(guild).onlinechannel_id.set(onlinechannel.id)
await self.update_infochannel(guild)
async def delete_all_infochannels(self, guild: discord.Guild):
guild_data = await self.config.guild(guild).all()
botchannel_id = guild_data["botchannel_id"]
onlinechannel_id = guild_data["onlinechannel_id"]
botchannel: discord.VoiceChannel = guild.get_channel(botchannel_id)
onlinechannel: discord.VoiceChannel = guild.get_channel(onlinechannel_id)
channel_id = guild_data["channel_id"]
channel: discord.VoiceChannel = guild.get_channel(channel_id)
await channel.delete(reason="InfoChannel delete")
if botchannel_id is not None:
await botchannel.delete(reason="InfoChannel delete")
if onlinechannel_id is not None:
await onlinechannel.delete(reason="InfoChannel delete")
await self.config.guild(guild).clear()
async def update_infochannel(self, guild: discord.Guild):
guild_data = await self.config.guild(guild).all()
botcount = guild_data["bot_count"]
onlinecount = guild_data["online_count"]
# Gets count of bots
# bots = lambda x: x.bot
# def bots(x): return x.bot
bot_num = len([m for m in guild.members if m.bot])
# bot_msg = f"Bots: {num}"
# Gets count of online users
members = guild.member_count
offline = len(list(filter(lambda m: m.status is discord.Status.offline, guild.members)))
online_num = members - offline
# online_msg = f"Online: {num}"
# Gets count of actual users
total = lambda x: not x.bot
human_num = len([m for m in guild.members if total(m)])
# human_msg = f"Total Humans: {num}"
channel_id = guild_data["channel_id"]
if channel_id is None:
return False
botchannel_id = guild_data["botchannel_id"]
onlinechannel_id = guild_data["onlinechannel_id"]
channel_id = guild_data["channel_id"]
channel: discord.VoiceChannel = guild.get_channel(channel_id)
botchannel: discord.VoiceChannel = guild.get_channel(botchannel_id)
onlinechannel: discord.VoiceChannel = guild.get_channel(onlinechannel_id)
if guild_data["member_count"]:
name = f"{channel.name.split(':')[0]}: {human_num}"
await channel.edit(reason="InfoChannel update", name=name)
if botcount:
name = f"{botchannel.name.split(':')[0]}: {bot_num}"
await botchannel.edit(reason="InfoChannel update", name=name)
if onlinecount:
name = f"{onlinechannel.name.split(':')[0]}: {online_num}"
await onlinechannel.edit(reason="InfoChannel update", name=name)
async def update_infochannel_with_cooldown(self, guild):
"""My attempt at preventing rate limits, lets see how it goes"""
if self._critical_section_wooah_:
if self._critical_section_wooah_ == 2:
# print("Already pending, skipping")
return # Another one is already pending, don't queue more than one
# print("Queuing another update")
self._critical_section_wooah_ = 2
while self._critical_section_wooah_:
await asyncio.sleep(
RATE_LIMIT_DELAY // 4
) # Max delay ends up as 1.25 * RATE_LIMIT_DELAY
# print("Issuing queued update")
return await self.update_infochannel_with_cooldown(guild)
# print("Entering critical")
self._critical_section_wooah_ = 1
await self.update_infochannel(guild)
await asyncio.sleep(RATE_LIMIT_DELAY)
self._critical_section_wooah_ = 0
# print("Exiting critical")
@Cog.listener()
async def on_member_join(self, member: discord.Member):
if await self.bot.cog_disabled_in_guild(self, member.guild):
return
await self.update_infochannel_with_cooldown(member.guild)
@Cog.listener()
async def on_member_remove(self, member: discord.Member):
if await self.bot.cog_disabled_in_guild(self, member.guild):
return
await self.update_infochannel_with_cooldown(member.guild)
@Cog.listener()
async def on_member_update(self, before: discord.Member, after: discord.Member):
if await self.bot.cog_disabled_in_guild(self, after.guild):
return
onlinecount = await self.config.guild(after.guild).online_count()
if onlinecount:
if before.status != after.status:
await self.update_infochannel_with_cooldown(after.guild)