248 lines
8.9 KiB
Python
248 lines
8.9 KiB
Python
import asyncio
|
|
import logging
|
|
from datetime import datetime, timedelta
|
|
|
|
import discord
|
|
from redbot.core import Config, checks, commands
|
|
from redbot.core.bot import Red
|
|
from redbot.core.commands import Cog, parse_timedelta
|
|
from redbot.core.utils import AsyncIter
|
|
from redbot.core.utils.chat_formatting import pagify
|
|
|
|
log = logging.getLogger("red.fox_v3.timerole")
|
|
|
|
|
|
async def sleep_till_next_hour():
|
|
now = datetime.utcnow()
|
|
next_hour = datetime(year=now.year, month=now.month, day=now.day, hour=now.hour + 1)
|
|
log.debug("Sleeping for {} seconds".format((next_hour - datetime.utcnow()).seconds))
|
|
await asyncio.sleep((next_hour - datetime.utcnow()).seconds)
|
|
|
|
|
|
class Timerole(Cog):
|
|
"""Add roles to users based on time on server"""
|
|
|
|
def __init__(self, bot: Red):
|
|
super().__init__()
|
|
self.bot = bot
|
|
self.config = Config.get_conf(self, identifier=9811198108111121, force_registration=True)
|
|
default_global = {}
|
|
default_guild = {"announce": None, "roles": {}}
|
|
|
|
self.config.register_global(**default_global)
|
|
self.config.register_guild(**default_guild)
|
|
self.updating = asyncio.create_task(self.check_hour())
|
|
|
|
async def red_delete_data_for_user(self, **kwargs):
|
|
"""Nothing to delete"""
|
|
return
|
|
|
|
def cog_unload(self):
|
|
self.updating.cancel()
|
|
|
|
@commands.command()
|
|
@checks.guildowner()
|
|
@commands.guild_only()
|
|
async def runtimerole(self, ctx: commands.Context):
|
|
"""
|
|
Trigger the hourly timerole
|
|
|
|
Useful for troubleshooting the initial setup
|
|
"""
|
|
|
|
async with ctx.typing():
|
|
await self.timerole_update()
|
|
await ctx.tick()
|
|
|
|
@commands.group()
|
|
@checks.mod_or_permissions(administrator=True)
|
|
@commands.guild_only()
|
|
async def timerole(self, ctx):
|
|
"""Adjust timerole settings"""
|
|
if ctx.invoked_subcommand is None:
|
|
pass
|
|
|
|
@timerole.command()
|
|
async def addrole(
|
|
self, ctx: commands.Context, role: discord.Role, time: str, *requiredroles: discord.Role
|
|
):
|
|
"""Add a role to be added after specified time on server"""
|
|
guild = ctx.guild
|
|
|
|
try:
|
|
parsed_time = parse_timedelta(time, allowed_units=["weeks", "days", "hours"])
|
|
except commands.BadArgument:
|
|
await ctx.maybe_send_embed("Error: Invalid time string.")
|
|
return
|
|
|
|
days = parsed_time.days
|
|
hours = parsed_time.seconds // 60 // 60
|
|
|
|
to_set = {"days": days, "hours": hours, "remove": False}
|
|
if requiredroles:
|
|
to_set["required"] = [r.id for r in requiredroles]
|
|
|
|
await self.config.guild(guild).roles.set_raw(role.id, value=to_set)
|
|
await ctx.maybe_send_embed(
|
|
"Time Role for {0} set to {1} days and {2} hours until added".format(
|
|
role.name, days, hours
|
|
)
|
|
)
|
|
|
|
@timerole.command()
|
|
async def removerole(
|
|
self, ctx: commands.Context, role: discord.Role, time: str, *requiredroles: discord.Role
|
|
):
|
|
"""
|
|
Add a role to be removed after specified time on server
|
|
|
|
Useful with an autorole cog
|
|
"""
|
|
guild = ctx.guild
|
|
try:
|
|
parsed_time = parse_timedelta(time, allowed_units=["weeks", "days", "hours"])
|
|
except commands.BadArgument:
|
|
await ctx.maybe_send_embed("Error: Invalid time string.")
|
|
return
|
|
|
|
days = parsed_time.days
|
|
hours = parsed_time.seconds // 60 // 60
|
|
|
|
to_set = {"days": days, "hours": hours, "remove": True}
|
|
if requiredroles:
|
|
to_set["required"] = [r.id for r in requiredroles]
|
|
|
|
await self.config.guild(guild).roles.set_raw(role.id, value=to_set)
|
|
await ctx.maybe_send_embed(
|
|
"Time Role for {0} set to {1} days and {2} hours until removed".format(
|
|
role.name, days, hours
|
|
)
|
|
)
|
|
|
|
@timerole.command()
|
|
async def channel(self, ctx: commands.Context, channel: discord.TextChannel):
|
|
"""Sets the announce channel for role adds"""
|
|
guild = ctx.guild
|
|
|
|
await self.config.guild(guild).announce.set(channel.id)
|
|
await ctx.send("Announce channel set to {0}".format(channel.mention))
|
|
|
|
@timerole.command()
|
|
async def delrole(self, ctx: commands.Context, role: discord.Role):
|
|
"""Deletes a role from being added/removed after specified time"""
|
|
guild = ctx.guild
|
|
|
|
await self.config.guild(guild).roles.set_raw(role.id, value=None)
|
|
await ctx.send("{0} will no longer be applied".format(role.name))
|
|
|
|
@timerole.command()
|
|
async def list(self, ctx: commands.Context):
|
|
"""Lists all currently setup timeroles"""
|
|
guild = ctx.guild
|
|
|
|
role_dict = await self.config.guild(guild).roles()
|
|
out = "Current Timeroles:\n"
|
|
for r_id, r_data in role_dict.items():
|
|
if r_data is not None:
|
|
role = discord.utils.get(guild.roles, id=int(r_id))
|
|
r_roles = []
|
|
if role is None:
|
|
role = r_id
|
|
if "required" in r_data:
|
|
r_roles = [
|
|
str(discord.utils.get(guild.roles, id=int(new_id)))
|
|
for new_id in r_data["required"]
|
|
]
|
|
out += "{} | {} days | requires: {}\n".format(str(role), r_data["days"], r_roles)
|
|
await ctx.maybe_send_embed(out)
|
|
|
|
async def timerole_update(self):
|
|
for guild in self.bot.guilds:
|
|
addlist = []
|
|
removelist = []
|
|
|
|
role_dict = await self.config.guild(guild).roles()
|
|
if not any(role_data for role_data in role_dict.values()): # No roles
|
|
continue
|
|
|
|
async for member in AsyncIter(guild.members):
|
|
has_roles = [r.id for r in member.roles]
|
|
|
|
add_roles = [
|
|
int(rID)
|
|
for rID, r_data in role_dict.items()
|
|
if r_data is not None and not r_data["remove"]
|
|
]
|
|
remove_roles = [
|
|
int(rID)
|
|
for rID, r_data in role_dict.items()
|
|
if r_data is not None and r_data["remove"]
|
|
]
|
|
|
|
check_add_roles = set(add_roles) - set(has_roles)
|
|
check_remove_roles = set(remove_roles) & set(has_roles)
|
|
|
|
await self.check_required_and_date(
|
|
addlist, check_add_roles, has_roles, member, role_dict
|
|
)
|
|
await self.check_required_and_date(
|
|
removelist, check_remove_roles, has_roles, member, role_dict
|
|
)
|
|
|
|
channel = await self.config.guild(guild).announce()
|
|
if channel is not None:
|
|
channel = guild.get_channel(channel)
|
|
|
|
title = "**These members have received the following roles**\n"
|
|
await self.announce_roles(title, addlist, channel, guild, to_add=True)
|
|
title = "**These members have lost the following roles**\n"
|
|
await self.announce_roles(title, removelist, channel, guild, to_add=False)
|
|
|
|
async def announce_roles(self, title, role_list, channel, guild, to_add: True):
|
|
results = ""
|
|
for member, role_id in role_list:
|
|
role = discord.utils.get(guild.roles, id=role_id)
|
|
try:
|
|
if to_add:
|
|
await member.add_roles(role, reason="Timerole")
|
|
else:
|
|
await member.remove_roles(role, reason="Timerole")
|
|
except (discord.Forbidden, discord.NotFound) as e:
|
|
results += "{} : {} **(Failed)**\n".format(member.display_name, role.name)
|
|
else:
|
|
results += "{} : {}\n".format(member.display_name, role.name)
|
|
if channel is not None and results:
|
|
await channel.send(title)
|
|
for page in pagify(results, shorten_by=50):
|
|
await channel.send(page)
|
|
elif results: # Channel is None, log the results
|
|
log.info(results)
|
|
|
|
async def check_required_and_date(self, role_list, check_roles, has_roles, member, role_dict):
|
|
for role_id in check_roles:
|
|
# Check for required role
|
|
if "required" in role_dict[str(role_id)]:
|
|
if not set(role_dict[str(role_id)]["required"]) & set(has_roles):
|
|
# Doesn't have required role
|
|
continue
|
|
|
|
if (
|
|
member.joined_at
|
|
+ timedelta(
|
|
days=role_dict[str(role_id)]["days"],
|
|
hours=role_dict[str(role_id)].get("hours", 0),
|
|
)
|
|
<= datetime.today()
|
|
):
|
|
# Qualifies
|
|
role_list.append((member, role_id))
|
|
|
|
async def check_hour(self):
|
|
await sleep_till_next_hour()
|
|
while self is self.bot.get_cog("Timerole"):
|
|
await self.timerole_update()
|
|
await sleep_till_next_hour()
|
|
|
|
|
|
|