@ -1,53 +1,25 @@
import asyncio import asyncio
import logging from typing import Union
from collections import defaultdict
from typing import Dict , Optional , Union
import discord import discord
from redbot . core import Config , checks , commands from redbot . core import Config , checks , commands
from redbot . core . bot import Red from redbot . core . bot import Red
from redbot . core . commands import Cog from redbot . core . commands import Cog
# 10 minutes. Rate limit is 2 per 10, so 1 per 6 is safe. # Cog: Any = getattr(commands, "Cog", object)
RATE_LIMIT_DELAY = 60 * 6 # If you're willing to risk rate limiting, you can decrease the delay # listener = getattr(commands.Cog, "listener", None) # Trusty + Sinbad
# if listener is None:
log = logging . getLogger ( " red.fox_v3.infochannel " ) # def listener(name=None):
# return lambda x: x
async def get_channel_counts ( category , guild ) : RATE_LIMIT_DELAY = 60 * 10 # If you're willing to risk rate limiting, you can decrease the delay
# Gets count of bots
bot_num = len ( [ m for m in guild . members if m . bot ] )
# Gets count of roles in the server
roles_num = len ( guild . roles ) - 1
# Gets count of channels in the server
# <number of total channels> - <number of channels in the stats category> - <categories>
channels_num = len ( guild . channels ) - len ( category . voice_channels ) - len ( guild . categories )
# Gets all counts of members
members = guild . member_count
offline_num = len ( list ( filter ( lambda m : m . status is discord . Status . offline , guild . members ) ) )
online_num = members - offline_num
# Gets count of actual users
human_num = members - bot_num
# count amount of premium subs/nitro subs.
boosters = guild . premium_subscription_count
return {
" members " : members ,
" humans " : human_num ,
" boosters " : boosters ,
" bots " : bot_num ,
" roles " : roles_num ,
" channels " : channels_num ,
" online " : online_num ,
" offline " : offline_num ,
}
class InfoChannel ( Cog ) : class InfoChannel ( Cog ) :
"""
"""
Create a channel with updating server info
Create a channel with updating server info
This relies on editing channels , which is a strictly rate - limited activity .
Less important information about the cog
As such , updates will not be frequent . Currently capped at 1 per 5 minutes per server .
"""
"""
def __init__ ( self , bot : Red ) :
def __init__ ( self , bot : Red ) :
@ -57,56 +29,23 @@ class InfoChannel(Cog):
self , identifier = 731101021116710497110110101108 , force_registration = True
self , identifier = 731101021116710497110110101108 , force_registration = True
)
)
# self. so I can get the keys from this later
self . default_channel_names = {
" members " : " Members: {count} " ,
" humans " : " Humans: {count} " ,
" boosters " : " Boosters: {count} " ,
" bots " : " Bots: {count} " ,
" roles " : " Roles: {count} " ,
" channels " : " Channels: {count} " ,
" online " : " Online: {count} " ,
" offline " : " Offline: {count} " ,
}
default_channel_ids = { k : None for k in self . default_channel_names }
# Only members is enabled by default
default_enabled_counts = { k : k == " members " for k in self . default_channel_names }
default_guild = {
default_guild = {
" category_id " : None ,
" channel_id " : None ,
" channel_ids " : default_channel_ids ,
" botchannel_id " : None ,
" enabled_channels " : default_enabled_counts ,
" onlinechannel_id " : None ,
" channel_names " : self . default_channel_names ,
" member_count " : True ,
" bot_count " : False ,
" online_count " : False ,
}
}
self . config . register_guild ( * * default_guild )
self . config . register_guild ( * * default_guild )
self . default_role = { " enabled " : False , " channel_id " : None , " name " : " {role} : {count} " }
self . config . register_role ( * * self . default_role )
self . _critical_section_wooah_ = 0
self . _critical_section_wooah_ = 0
self . channel_data = defaultdict ( dict )
self . edit_queue = defaultdict ( lambda : defaultdict ( lambda : asyncio . Queue ( maxsize = 2 ) ) )
self . _rate_limited_edits : Dict [ int , Dict [ str , Optional [ asyncio . Task ] ] ] = defaultdict (
lambda : defaultdict ( lambda : None )
)
async def red_delete_data_for_user ( self , * * kwargs ) :
async def red_delete_data_for_user ( self , * * kwargs ) :
""" Nothing to delete """
""" Nothing to delete """
return
return
async def initialize ( self ) :
for guild in self . bot . guilds :
await self . update_infochannel ( guild )
def cog_unload ( self ) :
self . stop_all_queues ( )
@commands.command ( )
@commands.command ( )
@checks.admin ( )
@checks.admin ( )
async def infochannel ( self , ctx : commands . Context ) :
async def infochannel ( self , ctx : commands . Context ) :
@ -122,466 +61,233 @@ class InfoChannel(Cog):
)
)
guild : discord . Guild = ctx . guild
guild : discord . Guild = ctx . guild
category_id = await self . config . guild ( guild ) . category_id ( )
channel_id = await self . config . guild ( guild ) . channel_id ( )
category = None
channel = None
if channel_id is not None :
if category_id is not None :
channel : Union [ discord . VoiceChannel , None ] = guild . get_channel ( channel_id )
category : Union [ discord . CategoryChannel , None ] = guild . get_channel ( category_id )
if c ategory_id is not None and category is None :
if c hannel_id is not None and channel is None :
await ctx . maybe_send_embed( " Info category has been deleted, recreate it?" )
await ctx . send( " Info channel has been deleted, recreate it?" )
elif c ategory _id is None :
elif c hannel _id is None :
await ctx . maybe_ send_embe d( " Enable info channel s on this server?" )
await ctx . ( " Enable info channel on this server?" )
else :
else :
await ctx . maybe_ send_embe d( " Do you wish to delete current info channels? " )
await ctx . ( " Do you wish to delete current info channels? " )
msg = await self . bot . wait_for ( " message " , check = check )
msg = await self . bot . wait_for ( " message " , check = check )
if msg . content . upper ( ) in [ " N " , " NO " ] :
if msg . content . upper ( ) in [ " N " , " NO " ] :
await ctx . maybe_ send_embe d( " Cancelled " )
await ctx . ( " Cancelled " )
return
return
if c ategory is None :
if c hannel is None :
try :
try :
await self . make_infochannel ( guild )
await self . make_infochannel ( guild )
except discord . Forbidden :
except discord . Forbidden :
await ctx . maybe_send_embed (
await ctx . send ( " Failure: Missing permission to create voice channel " )
" Failure: Missing permission to create necessary channels "
)
return
return
else :
else :
await self . delete_all_infochannels ( guild )
await self . delete_all_infochannels ( guild )
ctx . message = msg
if not await ctx . tick ( ) :
if not await ctx . tick ( ) :
await ctx . maybe_ send_embe d( " Done! " )
await ctx . send ( " Done! " )
@commands.group ( aliases = [ " icset " ] )
@commands.group ( )
@checks.admin ( )
@checks.admin ( )
async def infochannelset ( self , ctx : commands . Context ) :
async def infochannelset ( self , ctx : commands . Context ) :
"""
"""
Toggle different types of infochannels
Toggle different types of infochannels
"""
"""
pass
if not ctx . invoked_subcommand :
pass
@infochannelset.command ( name = " togglechannel " )
async def _infochannelset_togglechannel (
@infochannelset.command ( name = " botcount " )
self , ctx : commands . Context , channel_type : str , enabled : Optional [ bool ] = None
async def _infochannelset_botcount ( self , ctx : commands . Context , enabled : bool = None ) :
) :
"""
""" Toggles the infochannel for the specified channel type.
Toggle an infochannel that shows the amount of bots in the server
Valid Types are :
- ` members ` : Total members on the server
- ` humans ` : Total members that aren ' t bots
- ` boosters ` : Total amount of boosters
- ` bots ` : Total bots
- ` roles ` : Total number of roles
- ` channels ` : Total number of channels excluding infochannels ,
- ` online ` : Total online members ,
- ` offline ` : Total offline members ,
"""
"""
guild = ctx . guild
guild = ctx . guild
if channel_type not in self . default_channel_names . keys ( ) :
await ctx . maybe_send_embed ( " Invalid channel type provided. " )
return
if enabled is None :
enabled = not await self . config . guild ( guild ) . enabled_channels . get_raw ( channel_type )
await self . config . guild ( guild ) . enabled_channels . set_raw ( channel_type , value = enabled )
await self . make_infochannel ( ctx . guild , channel_type = channel_type )
if enabled :
await ctx . maybe_send_embed ( f " InfoChannel ` { channel_type } ` has been enabled. " )
else :
await ctx . maybe_send_embed ( f " InfoChannel ` { channel_type } ` has been disabled. " )
@infochannelset.command ( name = " togglerole " )
async def _infochannelset_rolecount (
self , ctx : commands . Context , role : discord . Role , enabled : bool = None
) :
""" Toggle an infochannel that shows the count of users with the specified role """
if enabled is None :
if enabled is None :
enabled = not await self . config . role( role ) . enabled ( )
enabled = not await self . config . guild ( guild ) . bot_count ( )
await self . config . role ( role ) . enabled . set ( enabled )
await self . config . guild ( guild ) . bot_count . set ( enabled )
await self . make_infochannel ( ctx . guild )
await self . make_infochannel ( ctx . guild , channel_role = role )
if enabled :
if enabled :
await ctx . maybe_send_embed( f " InfoChannel for { role . name } count has been enabled." )
await ctx . send ( " InfoChannel for bot count has been enabled. " )
else :
else :
await ctx . maybe_send_embed( f " InfoChannel for { role . name } count has been disabled." )
await ctx . send ( " InfoChannel for bot count has been disabled. " )
@infochannelset.command ( name = " name " )
@infochannelset.command ( name = " onlinecount " )
async def _infochannelset_ name ( self , ctx : commands . Context , channel_type: str , * , text = None ) :
async def _infochannelset_onlinecount ( self , ctx : commands . Context , enabled : bool = None ) :
"""
"""
Change the name of the infochannel for the specified channel type .
Toggle an infochannel that shows the amount of online users in the server
{ count } must be used to display number of total members in the server .
Leave blank to set back to default .
Examples :
- ` [ p ] infochannelset name members Cool Cats : { count } `
- ` [ p ] infochannelset name bots { count } Robot Overlords `
Valid Types are :
- ` members ` : Total members on the server
- ` humans ` : Total members that aren ' t bots
- ` boosters ` : Total amount of boosters
- ` bots ` : Total bots
- ` roles ` : Total number of roles
- ` channels ` : Total number of channels excluding infochannels
- ` online ` : Total online members
- ` offline ` : Total offline members
Warning : This command counts against the channel update rate limit and may be queued .
"""
"""
guild = ctx . guild
guild = ctx . guild
if channel_type not in self . default_channel_names . keys ( ) :
if enabled is None :
await ctx . maybe_send_embed ( " Invalid channel type provided. " )
enabled = not await self . config . guild ( guild ) . online_count ( )
return
if text is None :
text = self . default_channel_names . get ( channel_type )
elif " {count} " not in text :
await ctx . maybe_send_embed (
" Improperly formatted. Make sure to use ` {count} ` in your channel name "
)
return
elif len ( text ) > 93 :
await ctx . maybe_send_embed ( " Name is too long, max length is 93. " )
return
await self . config . guild ( guild ) . channel_names . set_raw ( channel_type , value = text )
await self . update_infochannel ( guild , channel_type = channel_type )
if not await ctx . tick ( ) :
await ctx . maybe_send_embed ( " Done! " )
@infochannelset.command ( name = " rolename " )
async def _infochannelset_rolename (
self , ctx : commands . Context , role : discord . Role , * , text = None
) :
"""
Change the name of the infochannel for specific roles .
{ count } must be used to display number members with the given role .
{ role } can be used for the roles name .
Leave blank to set back to default .
Default is set to : ` { role } : { count } `
Examples :
- ` [ p ] infochannelset rolename @Patrons { role } : { count } `
- ` [ p ] infochannelset rolename Elite { count } members with { role } role `
- ` [ p ] infochannelset rolename " Space Role " Total boosters : { count } `
Warning : This command counts against the channel update rate limit and may be queued .
"""
guild = ctx . message . guild
if text is None :
text = self . default_role [ " name " ]
elif " {count} " not in text :
await ctx . maybe_send_embed (
" Improperly formatted. Make sure to use ` {count} ` in your channel name "
)
return
await self . config . role ( role ) . name . set ( text )
await self . update_infochannel ( guild , channel_role = role )
if not await ctx . tick ( ) :
await ctx . maybe_send_embed ( " Done! " )
async def create_individual_channel (
self , guild , category : discord . CategoryChannel , overwrites , channel_type , count
) :
# Delete the channel if it exists
channel_id = await self . config . guild ( guild ) . channel_ids . get_raw ( channel_type )
if channel_id is not None :
channel : discord . VoiceChannel = guild . get_channel ( channel_id )
if channel :
self . stop_queue ( guild . id , channel_type )
await channel . delete ( reason = " InfoChannel delete " )
# Only make the channel if it's enabled
await self . config . guild ( guild ) . online_count . set ( enabled )
if await self . config . guild ( guild ) . enabled_channels . get_raw ( channel_type ) :
await self . make_infochannel ( ctx . guild )
name = await self . config . guild ( guild ) . channel_names . get_raw ( channel_type )
name = name . format ( count = count )
channel = await category . create_voice_channel (
name , reason = " InfoChannel make " , overwrites = overwrites
)
await self . config . guild ( guild ) . channel_ids . set_raw ( channel_type , value = channel . id )
return channel
return None
async def create_role_channel (
self , guild , category : discord . CategoryChannel , overwrites , role : discord . Role
) :
# Delete the channel if it exists
channel_id = await self . config . role ( role ) . channel_id ( )
if channel_id is not None :
channel : discord . VoiceChannel = guild . get_channel ( channel_id )
if channel :
self . stop_queue ( guild . id , role . id )
await channel . delete ( reason = " InfoChannel delete " )
# Only make the channel if it's enabled
if enabled :
if await self . config . role ( role ) . enabled ( ) :
await ctx . send ( " InfoChannel for online user count has been enabled. " )
count = len ( role . members )
else :
name = await self . config . role ( role ) . name ( )
await ctx . send ( " InfoChannel for online user count has been disabled. " )
name = name . format ( role = role . name , count = count )
channel = await category . create_voice_channel (
name , reason = " InfoChannel make " , overwrites = overwrites
)
await self . config . role ( role ) . channel_id . set ( channel . id )
return channel
return None
async def make_infochannel ( self , guild : discord . Guild , channel_type = None , channel_role = None ) :
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 = {
overwrites = {
guild . default_role : discord . PermissionOverwrite ( connect = False ) ,
guild . default_role : discord . PermissionOverwrite ( connect = False ) ,
guild . me : discord . PermissionOverwrite ( manage_channels = True , connect = True ) ,
guild . me : discord . PermissionOverwrite ( manage_channels = True , connect = True ) ,
}
}
# Check for and create the Infochannel category
# Remove the old info channel first
category_id = await self . config . guild ( guild ) . category_id ( )
channel_id = await self . config . guild ( guild ) . channel_id ( )
if category_id is not None :
if channel_id is not None :
category : discord . CategoryChannel = guild . get_channel ( category_id )
channel : discord . VoiceChannel = guild . get_channel ( channel_id )
if category is None : # Category id is invalid, probably deleted.
if channel :
category_id = None
await channel . delete ( reason = " InfoChannel delete " )
if category_id is None :
category : discord . CategoryChannel = await guild . create_category (
" Server Stats " , reason = " InfoChannel Category make "
)
await self . config . guild ( guild ) . category_id . set ( category . id )
await category . edit ( position = 0 )
category_id = category . id
category : discord . CategoryChannel = guild . get_channel ( category_id )
channel_data = await get_channel_counts ( category , guild )
# 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 )
# Only update a single channel
if botcount :
if channel_type is not None :
# Remove the old bot channel first
await self . create_individual_channel (
botchannel_id = await self . config . guild ( guild ) . botchannel_id ( )
guild , category , overwrites , channel_type , channel_data [ channel_type ]
if channel_id is not None :
)
botchannel : discord . VoiceChannel = guild . get_channel ( botchannel_id )
return
if botchannel :
if channel_role is not None :
await botchannel . delete ( reason = " InfoChannel delete " )
await self . create_role_channel ( guild , category , overwrites , channel_role )
return
# Update all channels
# Then create the new one
for channel_type in self . default_channel_names . keys ( ) :
botchannel = await guild . create_voice_channel (
await self . create_individual_channel (
" Bots: " , reason = " InfoChannel botcount " , overwrites = overwrites
guild , category , overwrites , channel_type , channel_data [ channel_type ]
)
)
await self . config . guild ( guild ) . botchannel_id . set ( botchannel . id )
for role in guild . roles :
if onlinecount :
await self . create_role_channel ( guild , category , overwrites , role )
# Remove the old online channel first
onlinechannel_id = await self . config . guild ( guild ) . onlinechannel_id ( )
# await self.update_infochannel(guild)
async def delete_all_infochannels ( self , guild : discord . Guild ) :
self . stop_guild_queues ( guild . id ) # Stop processing edits
# Delete regular channels
for channel_type in self . default_channel_names . keys ( ) :
channel_id = await self . config . guild ( guild ) . channel_ids . get_raw ( channel_type )
if channel_id is not None :
channel = guild . get_channel ( channel_id )
if channel is not None :
await channel . delete ( reason = " InfoChannel delete " )
await self . config . guild ( guild ) . channel_ids . clear_raw ( channel_type )
# Delete role channels
for role in guild . roles :
channel_id = await self . config . role ( role ) . channel_id ( )
if channel_id is not None :
if channel_id is not None :
channel = guild . get_channel ( channel_id )
onlinechannel : discord . VoiceChannel = guild . get_channel ( onlinechannel_id )
if channel is not None :
if onlinechannel :
await channel . delete ( reason = " InfoChannel delete " )
await onlinechannel . delete ( reason = " InfoChannel delete " )
await self . config . role ( role ) . channel_id . clear ( )
# Delete the category last
category_id = await self . config . guild ( guild ) . category_id ( )
if category_id is not None :
category = guild . get_channel ( category_id )
if category is not None :
await category . delete ( reason = " InfoChannel delete " )
async def add_to_queue ( self , guild , channel , identifier , count , formatted_name ) :
self . channel_data [ guild . id ] [ identifier ] = ( count , formatted_name , channel . id )
if not self . edit_queue [ guild . id ] [ identifier ] . full ( ) :
try :
self . edit_queue [ guild . id ] [ identifier ] . put_nowait ( identifier )
except asyncio . QueueFull :
pass # If queue is full, disregard
if self . _rate_limited_edits [ guild . id ] [ identifier ] is None :
await self . start_queue ( guild . id , identifier )
async def update_individual_channel ( self , guild , channel_type , count , guild_data ) :
# Then create the new one
name = guild_data [ " channel_names " ] [ channel_type ]
onlinechannel = await guild . create_voice_channel (
name = name . format ( count = count )
" Online: " , reason = " InfoChannel onlinecount " , overwrites = overwrites
channel = guild . get_channel ( guild_data [ " channel_ids " ] [ channel_type ] )
if channel is None :
return # abort
await self . add_to_queue ( guild , channel , channel_type , count , name )
async def update_role_channel ( self , guild , role : discord . Role , role_data ) :
if not role_data [ " enabled " ] :
return # Not enabled
count = len ( role . members )
name = role_data [ " name " ]
name = name . format ( role = role . name , count = count )
channel = guild . get_channel ( role_data [ " channel_id " ] )
if channel is None :
return # abort
await self . add_to_queue ( guild , channel , role . id , count , name )
async def update_infochannel ( self , guild : discord . Guild , channel_type = None , channel_role = None ) :
if channel_type is None and channel_role is None :
return await self . trigger_updates_for (
guild ,
members = True ,
humans = True ,
boosters = True ,
bots = True ,
roles = True ,
channels = True ,
online = True ,
offline = True ,
extra_roles = set ( guild . roles ) ,
)
)
await self . config . guild ( guild ) . onlinechannel_id . set ( onlinechannel . id )
if channel_type is not None :
await self . update_infochannel ( guild )
return await self . trigger_updates_for ( guild , * * { channel_type : True } )
return await self . trigger_updates_for ( guild , extra_roles = { channel_role } )
async def start_queue ( self , guild_id , identifier ) :
self . _rate_limited_edits [ guild_id ] [ identifier ] = asyncio . create_task (
self . _process_queue ( guild_id , identifier )
)
def stop_queue ( self , guild_id , identifier ) :
if self . _rate_limited_edits [ guild_id ] [ identifier ] is not None :
self . _rate_limited_edits [ guild_id ] [ identifier ] . cancel ( )
def stop_guild_queues ( self , guild_id ) :
for identifier in self . _rate_limited_edits [ guild_id ] . keys ( ) :
self . stop_queue ( guild_id , identifier )
def stop_all_queues ( self ) :
for guild_id in self . _rate_limited_edits . keys ( ) :
self . stop_guild_queues ( guild_id )
async def _process_queue ( self , guild_id , identifier ) :
async def delete_all_infochannels ( self , guild : discord . Guild ) :
while True :
identifier = await self . edit_queue [ guild_id ] [ identifier ] . get ( ) # Waits forever
count , formatted_name , channel_id = self . channel_data [ guild_id ] [ identifier ]
channel : discord . VoiceChannel = self . bot . get_channel ( channel_id )
if channel . name == formatted_name :
continue # Nothing to process
log . debug ( f " Processing guild_id: { guild_id } - identifier: { identifier } " )
try :
await channel . edit ( reason = " InfoChannel update " , name = formatted_name )
except ( discord . Forbidden , discord . HTTPException ) :
pass # Don't bother figuring it out
except discord . InvalidArgument :
log . exception ( f " Invalid formatted infochannel: { formatted_name } " )
else :
await asyncio . sleep ( RATE_LIMIT_DELAY ) # Wait a reasonable amount of time
async def trigger_updates_for ( self , guild , * * kwargs ) :
extra_roles : Optional [ set ] = kwargs . pop ( " extra_roles " , False )
guild_data = await self . config . guild ( guild ) . all ( )
guild_data = await self . config . guild ( guild ) . all ( )
botchannel_id = guild_data [ " botchannel_id " ]
to_update = (
onlinechannel_id = guild_data [ " onlinechannel_id " ]
kwargs . keys ( ) & [ key for key , value in guild_data [ " enabled_channels " ] . items ( ) if value ]
botchannel : discord . VoiceChannel = guild . get_channel ( botchannel_id )
) # Value in kwargs doesn't matter
onlinechannel : discord . VoiceChannel = guild . get_channel ( onlinechannel_id )
channel_id = guild_data [ " channel_id " ]
if to_update or extra_roles :
channel : discord . VoiceChannel = guild . get_channel ( channel_id )
log . debug ( f " { to_update =} \n "
await channel . delete ( reason = " InfoChannel delete " )
f " { extra_roles =} " )
if botchannel_id is not None :
await botchannel . delete ( reason = " InfoChannel delete " )
category = guild . get_channel ( guild_data [ " category_id " ] )
if onlinechannel_id is not None :
if category is None :
await onlinechannel . delete ( reason = " InfoChannel delete " )
log . debug ( ' Channel category is missing, updating must be off ' )
return # Nothing to update, must be off
await self . config . guild ( guild ) . clear ( )
channel_data = await get_channel_counts ( category , guild )
async def update_infochannel ( self , guild : discord . Guild ) :
if to_update :
guild_data = await self . config . guild ( guild ) . all ( )
for channel_type in to_update :
botcount = guild_data [ " bot_count " ]
await self . update_individual_channel (
onlinecount = guild_data [ " online_count " ]
guild , channel_type , channel_data [ channel_type ] , guild_data
)
# Gets count of bots
if extra_roles :
# bots = lambda x: x.bot
role_data = await self . config . all_roles ( )
# def bots(x): return x.bot
for channel_role in extra_roles :
if channel_role . id in role_data :
bot_num = len ( [ m for m in guild . members if m . bot ] )
await self . update_role_channel (
# bot_msg = f"Bots: {num}"
guild , channel_role , role_data [ channel_role . id ]
)
# Gets count of online users
members = guild . member_count
@Cog.listener ( name = " on_member_join " )
offline = len ( list ( filter ( lambda m : m . status is discord . Status . offline , guild . members ) ) )
@Cog.listener ( name = " on_member_remove " )
online_num = members - offline
async def on_member_join_remove ( self , member : discord . Member ) :
# online_msg = f"Online: {num}"
if await self . bot . cog_disabled_in_guild ( self , member . guild ) :
return
# Gets count of actual users
total = lambda x : not x . bot
if member . bot :
human_num = len ( [ m for m in guild . members if total ( m ) ] )
await self . trigger_updates_for (
# human_msg = f"Total Humans: {num}"
member . guild , members = True , bots = True , online = True , offline = True
)
channel_id = guild_data [ " channel_id " ]
else :
if channel_id is None :
await self . trigger_updates_for (
return False
member . guild , members = True , humans = True , online = True , offline = True
)
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 ( )
@Cog.listener ( )
async def on_member_update ( self , before : discord . Member , after : discord . Member ) :
async def on_member_join ( self , member : discord . Member ) :
if await self . bot . cog_disabled_in_guild ( self , after . guild ) :
if await self . bot . cog_disabled_in_guild ( self , member . guild ) :
return
if before . status != after . status :
return await self . trigger_updates_for ( after . guild , online = True , offline = True )
# XOR
c = set ( after . roles ) ^ set ( before . roles )
if c :
await self . trigger_updates_for ( after . guild , extra_roles = c )
@Cog.listener ( " on_guild_channel_create " )
@Cog.listener ( " on_guild_channel_delete " )
async def on_guild_channel_create_delete ( self , channel : discord . TextChannel ) :
if await self . bot . cog_disabled_in_guild ( self , channel . guild ) :
return
return
await self . trigger_updates_for ( channel . guild , channels = True )
await self . update_infochannel_with_cooldown ( member . guild )
@Cog.listener ( )
@Cog.listener ( )
async def on_guild_role_create ( self , role ) :
async def on_member_remove ( self , member : discord . Member ) :
if await self . bot . cog_disabled_in_guild ( self , role . guild ) :
if await self . bot . cog_disabled_in_guild ( self , member . guild ) :
return
return
await self . trigger_updates_for ( role . guild , roles = True )
await self . update_infochannel_with_cooldown ( member . guild )
@Cog.listener ( )
@Cog.listener ( )
async def on_guild_role_delete ( self , role ) :
async def on_member_update ( self , before : discord . Member , after : discord . Member ) :
if await self . bot . cog_disabled_in_guild ( self , role . guild ) :
if await self . bot . cog_disabled_in_guild ( self , after . guild ) :
return
return
await self . trigger_updates_for ( role . guild , roles = True )
onlinecount = await self . config . guild ( after . guild ) . online_count ( )
if onlinecount :
role_channel_id = await self . config . role ( role ) . channel_id ( )
if before . status != after . status :
if role_channel_id is not None :
await self . update_infochannel_with_cooldown ( after . guild )
rolechannel : discord . VoiceChannel = role . guild . get_channel ( role_channel_id )
if rolechannel :
await rolechannel . delete ( reason = " InfoChannel delete " )
await self . config . role ( role ) . clear ( )
bobloy