flifloo/Administrator-py
Archived
1
0
Fork 0
Code Activity

Merge pull request #56 from flifloo/slash

Browse source 
Slash
This commit is contained in:
Ethanell 2021-02-04 11:01:40 +01:00 committed by flifloo
commit 269020de56
19 changed files with 727 additions and 687 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
administrator/__init__.py
View file

@ -1,10 +1,12 @@
from discord import Intents
from discord_slash import SlashCommand
from administrator.config import config
import db
from discord.ext import commands
bot = commands.Bot(command_prefix=config.get("prefix"), intents=Intents.all())
slash = SlashCommand(bot, auto_register=True, auto_delete=True)
import extensions

72
administrator/check.py
View file

@ -1,4 +1,8 @@
import functools
from discord import Permissions
from discord.ext import commands
from discord.ext.commands import NoPrivateMessage, NotOwner, MissingPermissions
import db
@ -8,12 +12,62 @@ class ExtensionDisabled(commands.CheckFailure):
def is_enabled():
async def check(ctx: commands.Context):
if ctx.command.cog and ctx.guild:
s = db.Session()
es = s.query(db.ExtensionState).get((ctx.command.cog.qualified_name, ctx.guild.id))
s.close()
if es and not es.state:
raise ExtensionDisabled()
return True
return commands.check(check)
def check(func):
@functools.wraps(func)
async def wrapped(*args):
ctx = args[1]
if ctx.guild:
s = db.Session()
es = s.query(db.ExtensionState).get((args[0].qualified_name, ctx.guild.id))
s.close()
if es and not es.state:
return
# raise ExtensionDisabled()
return await func(*args)
return wrapped
return check
def is_owner():
def check(func):
@functools.wraps(func)
async def wrapped(*args):
ctx = args[1]
if not await ctx._discord.is_owner(ctx.author):
raise NotOwner('You do not own this bot.')
return await func(*args)
return wrapped
return check
def guild_only():
def check(func):
@functools.wraps(func)
async def wrapped(*args):
if args[1].guild is None:
raise NoPrivateMessage()
return await func(*args)
return wrapped
return check
def has_permissions(**perms):
invalid = set(perms) - set(Permissions.VALID_FLAGS)
if invalid:
raise TypeError('Invalid permission(s): %s' % (', '.join(invalid)))
def check(func):
@functools.wraps(func)
async def wrapped(*args):
ctx = args[1]
ch = ctx.channel
permissions = ch.permissions_for(ctx.author)
missing = [perm for perm, value in perms.items() if getattr(permissions, perm) != value]
if not missing or permissions.administrator:
return await func(*args)
raise MissingPermissions(missing)
return wrapped
return check

20
administrator/utils.py
View file

@ -1,12 +1,32 @@
import re
from datetime import timedelta
from discord import Message
from discord.ext.commands import BadArgument
from sqlalchemy.orm import Session
import db
msg_url_re = re.compile(r"^https://.*discord.*\.com/channels/[0-9]+/([0-9+]+)/([0-9]+)$")
async def get_message_by_url(ctx, url: str) -> Message:
r = msg_url_re.fullmatch(url)
if not r:
raise BadArgument()
r = r.groups()
c = ctx.guild.get_channel(int(r[0]))
if not c:
raise BadArgument()
m = await c.fetch_message(int(r[1]))
if not m or m.is_system():
raise BadArgument()
return m
def time_pars(s: str) -> timedelta:
match = re.fullmatch(r"(?:([0-9]+)W)*(?:([0-9]+)D)*(?:([0-9]+)H)*(?:([0-9]+)M)*(?:([0-9]+)S)*",
s.upper().replace(" ", "").strip())

100
extensions/extension.py
View file

@ -2,9 +2,13 @@ from traceback import format_exc
from discord.ext import commands
from discord import Embed, Guild
from discord.ext.commands import MissingPermissions, BadArgument, CommandError
from discord.ext.commands import BadArgument
from discord_slash import cog_ext, SlashContext, SlashCommandOptionType
from discord_slash.utils import manage_commands
import db
from administrator import slash
from administrator.check import has_permissions
from administrator.logger import logger
@ -15,69 +19,66 @@ logger = logger.getChild(extension_name)
class Extension(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
slash.get_cog_commands(self)
def description(self):
return "Manage bot's extensions"
@commands.group("extension", pass_context=True)
async def extension(self, ctx: commands.Context):
if ctx.invoked_subcommand is None:
await ctx.invoke(self.extension_help)
@extension.group("help", pass_context=True)
async def extension_help(self, ctx: commands.Context):
embed = Embed(title="Extension help")
for c, n, v in [[self.extension_list, "extension list", "List all enabled extensions"],
[self.extension_enable, "extension enable", "Enable an extensions"],
[self.extension_disable, "extension disable", "Disable an extensions"],
[self.extension_loaded, "extension loaded", "List all loaded extensions"],
[self.extension_load, "extension load <name>", "Load an extension"],
[self.extension_unload, "extension unload <name>", "Unload an extension"],
[self.extension_reload, "extension reload <name>", "Reload an extension"]]:
try:
if await c.can_run(ctx):
embed.add_field(name=n, value=v, inline=False)
except CommandError:
pass
if not embed.fields:
raise MissingPermissions("")
await ctx.send(embed=embed)
@extension.group("list", pass_context=True)
@commands.has_guild_permissions(administrator=True)
async def extension_list(self, ctx: commands.Context):
@cog_ext.cog_subcommand(base="extension", name="list", description="List all enabled extensions")
@has_permissions(administrator=True)
async def extension_list(self, ctx: SlashContext):
s = db.Session()
embed = Embed(title="Extensions list")
for es in s.query(db.ExtensionState).filter(db.ExtensionState.guild_id == ctx.guild.id):
embed.add_field(name=es.extension_name, value="Enable" if es.state else "Disable")
await ctx.send(embed=embed)
s.close()
await ctx.send(embeds=[embed])
@extension.group("enable", pass_context=True)
@commands.has_guild_permissions(administrator=True)
async def extension_enable(self, ctx: commands.Context, name: str):
@cog_ext.cog_subcommand(base="extension",
name="enable",
description="Enable an extensions",
options=[manage_commands.create_option("extension", "The extension to enable",
SlashCommandOptionType.STRING, True)])
@has_permissions(administrator=True)
async def extension_enable(self, ctx: SlashContext, name: str):
s = db.Session()
es = s.query(db.ExtensionState).get((name, ctx.guild.id))
if not es or es.state:
if not es:
raise BadArgument()
es.state = True
s.add(es)
s.commit()
s.close()
await ctx.message.add_reaction("\U0001f44d")
elif es.state:
message = "Extension already enabled"
else:
es.state = True
s.add(es)
s.commit()
s.close()
message = "\U0001f44d"
await ctx.send(content=message)
@extension.group("disable", pass_context=True)
@commands.has_guild_permissions(administrator=True)
async def extension_disable(self, ctx: commands.Context, name: str):
@cog_ext.cog_subcommand(base="extension",
name="disable",
description="Disable an extensions",
options=[manage_commands.create_option("extension", "The extension to disable",
SlashCommandOptionType.STRING, True)])
@has_permissions(administrator=True)
async def extension_disable(self, ctx: SlashContext, name: str):
s = db.Session()
es = s.query(db.ExtensionState).get((name, ctx.guild.id))
if not es or not es.state:
if not es:
raise BadArgument()
es.state = False
s.add(es)
s.commit()
s.close()
await ctx.message.add_reaction("\U0001f44d")
elif not es.state:
message = "Extension already disabled"
else:
es.state = False
s.add(es)
s.commit()
s.close()
message = "\U0001f44d"
await ctx.send(content=message)
@commands.group("extension", pass_context=True)
async def extension(self, ctx: commands.Context):
pass
@extension.group("loaded", pass_context=True)
@commands.is_owner()
@ -152,6 +153,9 @@ class Extension(commands.Cog):
s.commit()
s.close()
def cog_unload(self):
slash.remove_cog_commands(self)
def setup(bot):
logger.info(f"Loading...")

92
extensions/greetings.py
View file

@ -1,18 +1,14 @@
from discord.ext import commands
from discord import Member, Embed, Forbidden
from discord.ext.commands import BadArgument
from discord_slash import cog_ext, SlashContext, SlashCommandOptionType
from discord_slash.utils import manage_commands
from administrator.check import is_enabled
from administrator.check import is_enabled, guild_only, has_permissions
from administrator.logger import logger
from administrator import db, config
from administrator import db, slash
from administrator.utils import event_is_enabled
def check_greetings_message_type(message_type):
if message_type not in ["join", "leave"]:
raise BadArgument()
extension_name = "greetings"
logger = logger.getChild(extension_name)
@ -20,68 +16,75 @@ logger = logger.getChild(extension_name)
class Greetings(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
slash.get_cog_commands(self)
def description(self):
return "Setup join and leave message"
@commands.group("greetings", pass_context=True)
@cog_ext.cog_subcommand(base="greetings", name="set",
description="Set the greetings message\n`{}` will be replace by the username",
options=[
manage_commands.create_option("type", "The join or leave message",
SlashCommandOptionType.STRING, True,
[manage_commands.create_choice("join", "join"),
manage_commands.create_choice("leave", "leave")]),
manage_commands.create_option("message", "The message", SlashCommandOptionType.STRING,
True)
])
@is_enabled()
@commands.guild_only()
@commands.has_permissions(manage_guild=True)
async def greetings(self, ctx: commands.Context):
if ctx.invoked_subcommand is None:
await ctx.invoke(self.greetings_help)
@greetings.group("help", pass_context=True)
async def greetings_help(self, ctx: commands.Context):
embed = Embed(title="Greetings help")
embed.add_field(name="set <join/leave> <message>", value="Set the greetings message\n"
"`{}` will be replace by the username",
inline=False)
embed.add_field(name="show <join/leave>", value="Show the greetings message", inline=False)
embed.add_field(name="toggle <join/leave>", value="Enable or disable the greetings message", inline=False)
await ctx.send(embed=embed)
@greetings.group("set", pass_context=True)
async def greetings_set(self, ctx: commands.Context, message_type: str):
check_greetings_message_type(message_type)
message = ctx.message.content.replace(config.get("prefix")+"greetings set " + message_type, "").strip()
@guild_only()
@has_permissions(manage_guild=True)
async def greetings_set(self, ctx: SlashContext, message_type: str, message: str):
s = db.Session()
m = s.query(db.Greetings).filter(db.Greetings.guild == ctx.guild.id).first()
if not m:
m = db.Greetings(ctx.guild.id)
s.add(m)
setattr(m, message_type+"_enable", True)
setattr(m, message_type+"_message", message)
setattr(m, message_type+"_message", message.replace("\\n", '\n'))
s.commit()
await ctx.message.add_reaction("\U0001f44d")
await ctx.send(content="\U0001f44d")
@greetings.group("show", pass_context=True)
async def greetings_show(self, ctx: commands.Context, message_type: str):
check_greetings_message_type(message_type)
@cog_ext.cog_subcommand(base="greetings", name="show",
description="Show the greetings message",
options=[manage_commands.create_option("type", "The join or leave message",
SlashCommandOptionType.STRING, True,
[manage_commands.create_choice("join", "join"),
manage_commands.create_choice("leave", "leave")])])
@is_enabled()
@guild_only()
@has_permissions(manage_guild=True)
async def greetings_show(self, ctx: SlashContext, message_type: str):
s = db.Session()
m = s.query(db.Greetings).filter(db.Greetings.guild == ctx.guild.id).first()
s.close()
if not m:
await ctx.send(f"No {message_type} message set !")
await ctx.send(content=f"No {message_type} message set !")
else:
if message_type == "join":
await ctx.send(embed=m.join_embed(ctx.guild.name, str(ctx.message.author)))
await ctx.send(embeds=[m.join_embed(ctx.guild.name, str(ctx.author))])
else:
await ctx.send(m.leave_msg(str(ctx.message.author)))
await ctx.send(content=m.leave_msg(str(ctx.author)))
@greetings.group("toggle", pass_context=True)
async def greetings_toggle(self, ctx: commands.Context, message_type: str):
check_greetings_message_type(message_type)
@cog_ext.cog_subcommand(base="greetings", name="toggle",
description="Enable or disable the greetings message",
options=[manage_commands.create_option("type", "The join or leave message",
SlashCommandOptionType.STRING, True,
[manage_commands.create_choice("join", "join"),
manage_commands.create_choice("leave", "leave")])])
@is_enabled()
@guild_only()
@has_permissions(manage_guild=True)
async def greetings_toggle(self, ctx: SlashContext, message_type: str):
s = db.Session()
m = s.query(db.Greetings).filter(db.Greetings.guild == ctx.guild.id).first()
if not m:
await ctx.send(f"No {message_type} message set !")
await ctx.send(content=f"No {message_type} message set !")
else:
setattr(m, message_type+"_enable", not getattr(m, message_type+"_enable"))
s.commit()
await ctx.send(f"{message_type.title()} message is " +
("enable" if getattr(m, message_type+"_enable") else "disable"))
await ctx.send(content=f"{message_type.title()} message is " +
("enable" if getattr(m, message_type+"_enable") else "disable"))
s.close()
@commands.Cog.listener()
@ -108,6 +111,9 @@ class Greetings(commands.Cog):
if m and m.leave_enable:
await member.guild.system_channel.send(m.leave_msg(str(member)))
def cog_unload(self):
slash.remove_cog_commands(self)
def setup(bot):
logger.info(f"Loading...")

50
extensions/help.py
View file

@ -1,9 +1,8 @@
from discord import Embed
from discord.ext import commands
from discord.ext.commands import CommandNotFound, MissingRequiredArgument, BadArgument, MissingPermissions, \
NoPrivateMessage, CommandError, NotOwner
NoPrivateMessage, NotOwner
from discord_slash import SlashContext
from administrator import config
from administrator.check import ExtensionDisabled
from administrator.logger import logger
@ -19,38 +18,35 @@ class Help(commands.Cog):
def description(self):
return "Give help and command list"
@commands.command("help", pass_context=True)
async def help(self, ctx: commands.Context):
embed = Embed(title="Help")
for c in filter(lambda x: x != "Help", self.bot.cogs):
cog = self.bot.cogs[c]
try:
if await getattr(cog, c.lower()).can_run(ctx):
embed.add_field(name=c,
value=cog.description() + "\n" +
f"`{config.get('prefix')}{c.lower()} help` for more information",
inline=False)
except CommandError:
pass
await ctx.send(embed=embed)
@commands.Cog.listener()
async def on_command_error(self, ctx: commands.Context, error):
await self.error_handler(ctx, error)
@commands.Cog.listener()
async def on_slash_command_error(self, ctx: SlashContext, error: Exception):
await self.error_handler(ctx, error)
async def error_handler(self, ctx, error: Exception):
if isinstance(error, CommandNotFound):
await ctx.message.add_reaction("\u2753")
await self.reaction(ctx, "\u2753")
elif isinstance(error, MissingRequiredArgument) or isinstance(error, BadArgument):
await ctx.message.add_reaction("\u274C")
elif isinstance(error, NotOwner) or isinstance(error, MissingPermissions)\
await self.reaction(ctx, "\u274C")
elif isinstance(error, NotOwner) or isinstance(error, MissingPermissions) \
or isinstance(error, NoPrivateMessage):
await ctx.message.add_reaction("\U000026D4")
await self.reaction(ctx, "\U000026D4")
elif isinstance(error, ExtensionDisabled):
await ctx.message.add_reaction("\U0001F6AB")
await self.reaction(ctx, "\U0001F6AB")
else:
await ctx.send("An error occurred !")
await ctx.send(content="An error occurred !")
raise error
await ctx.message.delete(delay=30)
@staticmethod
async def reaction(ctx, react: str):
m = getattr(ctx, "message", None)
if m:
await m.add_reaction(react)
else:
await ctx.send(content=react)
def setup(bot):

48
extensions/invite.py
View file

@ -1,12 +1,16 @@
import re
from discord import Embed, Member, Guild
from discord import Embed, Member, Guild, Role, CategoryChannel
from discord.abc import GuildChannel
from discord.errors import Forbidden
from discord.ext import commands
from discord.ext.commands import BadArgument
from discord_slash import cog_ext, SlashContext, SlashCommandOptionType
from discord_slash.utils import manage_commands
import db
from administrator.check import is_enabled
from administrator import slash
from administrator.check import is_enabled, guild_only, has_permissions
from administrator.logger import logger
from administrator.utils import event_is_enabled
@ -20,41 +24,31 @@ class Invite(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
self.invites = {}
slash.get_cog_commands(self)
self.bot.loop.create_task(self.update_invites())
def description(self):
return "Get role from a special invite link"
@commands.group("invite", pass_context=True)
@is_enabled()
@commands.guild_only()
@commands.has_guild_permissions(administrator=True)
async def invite(self, ctx: commands.Context):
if ctx.invoked_subcommand is None:
await ctx.invoke(self.invite_help)
@invite.group("help", pass_context=True)
async def invite_help(self, ctx: commands.Context):
embed = Embed(title="Invite help")
embed.add_field(name="invite create <#channel> <@role>", value="Create a invite link to a role", inline=False)
embed.add_field(name="invite delete <code>", value="Remove a invite", inline=False)
await ctx.send(embed=embed)
@invite.group("create", pass_context=True)
async def invite_add(self, ctx: commands.Context, channel: str, role: str):
if not channel_mention_re.fullmatch(channel) or len(ctx.message.channel_mentions) != 1 or\
not role_mention_re.fullmatch(role) or len(ctx.message.role_mentions) != 1:
@guild_only()
@has_permissions(administrator=True)
async def invite_add(self, ctx: SlashContext, channel: GuildChannel, role: Role):
if isinstance(channel, CategoryChannel):
raise BadArgument()
inv = await ctx.message.channel_mentions[0].create_invite()
inv = await channel.create_invite()
s = db.Session()
s.add(db.InviteRole(ctx.guild.id, inv.code, ctx.message.role_mentions[0].id))
s.add(db.InviteRole(ctx.guild.id, inv.code, role.id))
s.commit()
s.close()
await ctx.send(f"Invite created: `{inv.url}`")
await ctx.send(content=f"Invite created: `{inv.url}`")
@invite.group("delete", pass_context=True)
async def invite_delete(self, ctx: commands.Context, code: str):
@cog_ext.cog_subcommand(base="invite", name="delete", description="Remove a invite", options=[
manage_commands.create_option("code", "The invitation code", SlashCommandOptionType.STRING, True)])
@is_enabled()
@guild_only()
@has_permissions(administrator=True)
async def invite_delete(self, ctx: SlashContext, code: str):
inv = next(filter(lambda i: i.code == code, await ctx.guild.invites()), None)
if not inv:
raise BadArgument()
@ -68,7 +62,7 @@ class Invite(commands.Cog):
s.commit()
s.close()
await inv.delete()
await ctx.message.add_reaction("\U0001f44d")
await ctx.send(content="\U0001f44d")
async def update_invites(self):
for g in self.bot.guilds:

252
extensions/pcp.py
View file

@ -1,105 +1,74 @@
import re
from discord import Embed, Member
from discord import Member, Role
from discord.ext import commands
from discord.ext.commands import BadArgument, MissingPermissions, CommandError
from discord.ext.commands import BadArgument
from discord_slash import cog_ext, SlashCommandOptionType, SlashContext
from discord_slash.utils import manage_commands
import db
from administrator import slash
from administrator.check import guild_only, has_permissions
from administrator.logger import logger
from administrator.utils import get_message_by_url
extension_name = "PCP"
logger = logger.getChild(extension_name)
msg_url_re = re.compile(r"^https://.*discord.*\.com/channels/[0-9]+/([0-9+]+)/([0-9]+)$")
role_mention_re = re.compile(r"^<@&[0-9]+>$")
user_mention_re = re.compile(r"^<@![0-9]+>$")
class PCP(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
slash.get_cog_commands(self)
def description(self):
return "PCP Univ Lyon 1"
@commands.group("pcp", pass_context=True)
@commands.guild_only()
async def pcp(self, ctx: commands.Context):
group = ctx.message.content.replace(f"{ctx.prefix}{ctx.command} ", "").upper()
if group:
s = db.Session()
p = s.query(db.PCP).get(ctx.guild.id)
s.close()
if p and re.fullmatch(p.roles_re, group):
await ctx.message.add_reaction("\U000023f3")
role = next(filter(lambda r: r.name.upper() == group, ctx.guild.roles), None)
def roles() -> list:
return list(filter(
lambda r: re.fullmatch(p.roles_re, r.name.upper()) or
(p.start_role_re and re.fullmatch(p.start_role_re, r.name.upper())),
ctx.author.roles
))
if not role or role.name in map(lambda r: r.name, roles()):
await ctx.message.remove_reaction("\U000023f3", self.bot.user)
raise BadArgument()
while roles():
await ctx.author.remove_roles(*roles())
while role not in ctx.author.roles:
await ctx.author.add_roles(role)
await ctx.message.remove_reaction("\U000023f3", self.bot.user)
await ctx.message.add_reaction("\U0001f44d")
return
if ctx.invoked_subcommand is None:
await ctx.invoke(self.pcp_help)
@pcp.group("help", pass_context=True)
async def pcp_help(self, ctx: commands.Context):
embed = Embed(title="PCP help")
@cog_ext.cog_subcommand(base="pcp", name="join", description="Join your group", options=[
manage_commands.create_option("group", "The target group to join", SlashCommandOptionType.ROLE, True)])
@guild_only()
async def pcp(self, ctx: SlashContext, role: Role):
s = db.Session()
p = s.query(db.PCP).get(ctx.guild.id)
s.close()
if p:
embed.add_field(name="pcp <group>", value="Join your group", inline=False)
for c, n, v in [[self.pcp_group, "pcp group", "Manage PCP group"],
[self.pcp_pin, "pcp pin <url>", "Pin a message with the url"],
[self.pcp_unpin, "pcp unpin <url>", "Unpin a message with the url"]]:
try:
if await c.can_run(ctx):
embed.add_field(name=n, value=v, inline=False)
except CommandError:
pass
if not embed.fields:
raise MissingPermissions("")
await ctx.send(embed=embed)
if p and re.fullmatch(p.roles_re, role.name.upper()):
await ctx.send(content="\U000023f3")
@pcp.group("pin", pass_context=True)
async def pcp_pin(self, ctx: commands.Context, url: str):
async def roles() -> list:
return list(filter(
lambda r: re.fullmatch(p.roles_re, r.name.upper()) or
(p.start_role_re and re.fullmatch(p.start_role_re, r.name.upper())),
(await ctx.guild.fetch_member(ctx.author.id)).roles
))
if not role or role.name in map(lambda r: r.name, await roles()):
await ctx.delete()
raise BadArgument()
while await roles():
await ctx.author.remove_roles(*(await roles()))
while role not in (await ctx.guild.fetch_member(ctx.author.id)).roles:
await ctx.author.add_roles(role)
await ctx.edit(content="\U0001f44d")
@cog_ext.cog_subcommand(base="pcp", name="pin", description="Pin a message with the url", options=[
manage_commands.create_option("url", "message URL", SlashCommandOptionType.STRING, True)
])
@guild_only()
async def pcp_pin(self, ctx: SlashContext, url: str):
await self.pin(ctx, url, True)
@pcp.group("unpin", pass_context=True)
async def pcp_unpin(self, ctx: commands.Context, url: str):
@cog_ext.cog_subcommand(base="pcp", name="unpin", description="Unpin a message with the url", options=[
manage_commands.create_option("url", "message URL", SlashCommandOptionType.STRING, True)
])
@guild_only()
async def pcp_unpin(self, ctx: SlashContext, url: str):
await self.pin(ctx, url, False)
@staticmethod
async def pin(ctx: commands.Context, url: str, action: bool):
r = msg_url_re.fullmatch(url)
if not r:
raise BadArgument()
r = r.groups()
c = ctx.guild.get_channel(int(r[0]))
if not c:
raise BadArgument()
m = await c.fetch_message(int(r[1]))
if not m:
raise BadArgument()
async def pin(ctx: SlashContext, url: str, action: bool):
m = await get_message_by_url(ctx, url)
if action:
await m.pin()
msg = "pinned a message"
@ -107,35 +76,23 @@ class PCP(commands.Cog):
await m.unpin()
msg = "unpinned a message"
await ctx.send(f"{ctx.author.mention} {msg}")
await ctx.send(content=f"{ctx.author.mention} {msg}")
@pcp.group("group", pass_context=True)
@commands.has_permissions(administrator=True)
async def pcp_group(self, ctx: commands.Context):
if ctx.invoked_subcommand is None:
await ctx.invoke(self.pcp_group_help)
@pcp_group.group("help", pass_context=True)
async def pcp_group_help(self, ctx: commands.Context):
embed = Embed(title="PCP group help")
embed.add_field(name="pcp group set <role Regex> [Welcome role Regex]",
value="Set regex for group role", inline=False)
embed.add_field(name="pcp group unset", value="Unset regex for group role", inline=False)
embed.add_field(name="pcp group subject", value="Manage subjects for group", inline=False)
embed.add_field(name="pcp group fix_vocal",
value="Check all text channel permissions to reapply vocal permissions", inline=False)
await ctx.send(embed=embed)
@pcp_group.group("fix_vocal", pass_context=True)
async def pcp_group_fix_vocal(self, ctx: commands.Context):
@cog_ext.cog_subcommand(base="pcp", subcommand_group="group", name="fix_vocal",
description="Check all text channel permissions to reapply vocal permissions")
@has_permissions(administrator=True)
async def pcp_group_fix_vocal(self, ctx: SlashContext):
s = db.Session()
p = s.query(db.PCP).get(ctx.guild.id)
s.close()
if not p:
raise BadArgument()
message = "\U000023f3"
await ctx.send(content=message)
for cat in filter(lambda c: re.fullmatch(p.roles_re, c.name.upper()), ctx.guild.categories):
await ctx.send(f"{cat.name}...")
message += f"\n{cat.name}..."
await ctx.edit(content=message)
teachers = []
for t in cat.text_channels:
for p in t.overwrites:
@ -144,11 +101,20 @@ class PCP(commands.Cog):
voc = next(filter(lambda c: c.name == "vocal-1", cat.voice_channels), None)
for t in teachers:
await voc.set_permissions(t, view_channel=True)
await ctx.send(f"{cat.name} done")
await ctx.message.add_reaction("\U0001f44d")
message += f"\n{cat.name} done"
await ctx.edit(content=message)
message += "\n\U0001f44d"
await ctx.edit(content=message)
@pcp_group.group("set", pass_context=True)
async def pcp_group_set(self, ctx: commands.Context, roles_re: str, start_role_re: str = None):
@cog_ext.cog_subcommand(base="pcp", subcommand_group="group", name="set", description="Set regex for group role",
options=[
manage_commands.create_option("role", "Roles regex",
SlashCommandOptionType.STRING, True),
manage_commands.create_option("role2", "Start roles regex",
SlashCommandOptionType.STRING, False)
])
@has_permissions(administrator=True)
async def pcp_group_set(self, ctx: SlashContext, roles_re: str, start_role_re: str = None):
s = db.Session()
p = s.query(db.PCP).get(ctx.guild.id)
if p:
@ -159,9 +125,11 @@ class PCP(commands.Cog):
s.add(p)
s.commit()
s.close()
await ctx.message.add_reaction("\U0001f44d")
await ctx.send(content="\U0001f44d")
@pcp_group.group("unset", pass_context=True)
@cog_ext.cog_subcommand(base="pcp", subcommand_group="group", name="unset",
description="Unset regex for group role")
@has_permissions(administrator=True)
async def pcp_group_unset(self, ctx: commands.Context):
s = db.Session()
p = s.query(db.PCP).get(ctx.guild.id)
@ -173,33 +141,21 @@ class PCP(commands.Cog):
s.close()
await ctx.message.add_reaction("\U0001f44d")
@pcp_group.group("subject", pass_context=True)
async def pcp_group_subject(self, ctx: commands.Context):
if ctx.invoked_subcommand is None:
await ctx.invoke(self.pcp_group_subject_help)
@pcp_group_subject.group("help", pass_context=True)
async def pcp_group_subject_help(self, ctx: commands.Context):
embed = Embed(title="PCP group subject help")
embed.add_field(name="pcp group subject add <name> <@group> [@teacher]", value="Add a subject to a group",
inline=False)
embed.add_field(name="pcp group subject bulk <@group> [subject1] [subject2] ...", value="Bulk subject add",
inline=False)
embed.add_field(name="pcp group subject remove <name> <@group>", value="Remove a subject to a group",
inline=False)
await ctx.send(embed=embed)
@pcp_group_subject.group("add", pass_context=True)
async def pcp_group_subject_add(self, ctx: commands.Context, name: str, group: str, teacher: str = None):
if not role_mention_re.fullmatch(group):
raise BadArgument()
if teacher and not user_mention_re.fullmatch(teacher):
raise BadArgument()
elif teacher and\
not next(filter(lambda r: r.name == "professeurs", ctx.message.mentions[0].roles), None):
@cog_ext.cog_subcommand(base="pcp", subcommand_group="subject", name="add", description="Add a subject to a group",
options=[
manage_commands.create_option("name", "The subject name",
SlashCommandOptionType.STRING, True),
manage_commands.create_option("group", "The group",
SlashCommandOptionType.ROLE, True),
manage_commands.create_option("teacher", "The teacher",
SlashCommandOptionType.USER, False)
])
@has_permissions(administrator=True)
async def pcp_group_subject_add(self, ctx: SlashContext, name: str, group: Role, teacher: Member = None):
if teacher and not next(filter(lambda r: r.name == "professeurs", teacher.roles), None):
raise BadArgument()
cat = next(filter(lambda c: c.name.upper() == ctx.message.role_mentions[0].name.upper(),
cat = next(filter(lambda c: c.name.upper() == group.name.upper(),
ctx.guild.categories), None)
if not cat:
raise BadArgument()
@ -210,26 +166,32 @@ class PCP(commands.Cog):
voc = next(filter(lambda c: c.name == "vocal-1", cat.voice_channels), None)
if not voc:
voc = await cat.create_voice_channel("vocal-1")
if ctx.message.mentions:
await chan.set_permissions(ctx.message.mentions[0], read_messages=True)
await voc.set_permissions(ctx.message.mentions[0], view_channel=True)
if teacher:
await chan.set_permissions(teacher, read_messages=True)
await voc.set_permissions(teacher, view_channel=True)
await ctx.message.add_reaction("\U0001f44d")
await ctx.send(content="\U0001f44d")
@pcp_group_subject.group("bulk", pass_context=True)
async def pcp_group_subject_bulk(self, ctx: commands.Context, mention, *names):
if not role_mention_re.fullmatch(mention):
raise BadArgument()
for n in names:
await ctx.invoke(self.pcp_group_subject_add, n, mention)
@cog_ext.cog_subcommand(base="pcp", subcommand_group="subject", name="bulk",
description="Remove a subject to a group", options=[
manage_commands.create_option("group", "The group", SlashCommandOptionType.ROLE, True),
manage_commands.create_option("names", "Subjects names", SlashCommandOptionType.STRING, True)
])
@has_permissions(administrator=True)
async def pcp_group_subject_bulk(self, ctx: SlashContext, group: Role, names: str):
for n in names.split(" "):
await self.pcp_group_subject_add.invoke(ctx, n, group)
@pcp_group_subject.group("remove", pass_context=True)
async def pcp_group_subject_remove(self, ctx: commands.Context, name: str, group: str):
if not role_mention_re.fullmatch(group):
raise BadArgument()
cat = next(filter(lambda c: c.name.upper() == ctx.message.role_mentions[0].name.upper(),
ctx.guild.categories), None)
@cog_ext.cog_subcommand(base="pcp", subcommand_group="subject", name="remove",description="Bulk subject add",
options=[
manage_commands.create_option("name", "The subject name",
SlashCommandOptionType.STRING, True),
manage_commands.create_option("group", "The group",
SlashCommandOptionType.ROLE, True)
])
@has_permissions(administrator=True)
async def pcp_group_subject_remove(self, ctx: SlashContext, name: str, group: Role):
cat = next(filter(lambda c: c.name.upper() == group.name.upper(), ctx.guild.categories), None)
if not cat:
raise BadArgument()
@ -239,7 +201,7 @@ class PCP(commands.Cog):
await chan.delete()
await ctx.message.add_reaction("\U0001f44d")
await ctx.send(content="\U0001f44d")
def setup(bot):

70
extensions/poll.py
View file

@ -1,12 +1,16 @@
import shlex
from datetime import datetime
from discord.abc import GuildChannel
from discord.ext import commands
from discord import Embed, RawReactionActionEvent, RawMessageDeleteEvent, RawBulkMessageDeleteEvent, TextChannel, Guild
from discord.ext.commands import BadArgument
from discord_slash import cog_ext, SlashCommandOptionType, SlashContext
from discord_slash.utils import manage_commands
import db
from administrator.check import is_enabled
from administrator import slash
from administrator.check import is_enabled, guild_only
from administrator.logger import logger
from administrator.utils import event_is_enabled
@ -21,47 +25,41 @@ REACTIONS.append("\U0001F51F")
class Poll(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
slash.get_cog_commands(self)
def description(self):
return "Create poll with a simple command"
@commands.group("poll", pass_context=True)
@cog_ext.cog_slash(name="poll",
description="Create a poll",
options=[
manage_commands.create_option("name", "Poll name",
SlashCommandOptionType.STRING, True),
manage_commands.create_option("choices", "All pool choice",
SlashCommandOptionType.STRING, True),
manage_commands.create_option("multi", "Allow multiple choice",
SlashCommandOptionType.BOOLEAN, False)
])
@is_enabled()
@commands.guild_only()
async def poll(self, ctx: commands.Context, name: str, *choices):
if name == "help":
await ctx.invoke(self.poll_help)
@guild_only()
async def poll(self, ctx: SlashContext, name: str, choices: str, multi: bool = False):
choices = shlex.split(choices)
if len(choices) > 11:
raise BadArgument()
else:
multi = False
if choices and choices[0] in ["multi", "m"]:
multi = True
choices = choices[1:]
if len(choices) == 0 or len(choices) > 11:
raise BadArgument()
else:
embed = Embed(title=f"Poll: {name}")
embed.set_author(name=str(ctx.author), icon_url=ctx.author.avatar_url)
embed.set_footer(text=f"Created: {ctx.message.created_at.strftime('%d/%m/%Y %H:%M')}")
for i, choice in enumerate(choices):
embed.add_field(name=REACTIONS[i], value=choice, inline=False)
message = await ctx.send(embed=embed)
reactions = REACTIONS[0:len(choices)] + ["\U0001F5D1"]
for reaction in reactions:
await message.add_reaction(reaction)
s = db.Session()
s.add(db.Polls(message.id, ctx.channel.id, ctx.guild.id, ctx.message.author.id, reactions, multi))
s.commit()
s.close()
await ctx.message.delete()
@poll.group("help", pass_context=True)
async def poll_help(self, ctx: commands.Context):
embed = Embed(title="Poll help")
embed.add_field(name="poll <name> [multi|m] <Choice N°1> <Choice N°2> ... <Choice N°11>",
value="Create a poll, the argument multi (or m) after the name allow multiple response\n"
"User the \U0001F5D1 to close the poll",
inline=False)
await ctx.send(embed=embed)
embed = Embed(title=f"Poll: {name}")
embed.set_author(name=str(ctx.author), icon_url=ctx.author.avatar_url)
embed.set_footer(text=f"Created: {datetime.now().strftime('%d/%m/%Y %H:%M')}")
for i, choice in enumerate(choices):
embed.add_field(name=REACTIONS[i], value=choice, inline=False)
message = await ctx.channel.send(embed=embed)
reactions = REACTIONS[0:len(choices)] + ["\U0001F5D1"]
for reaction in reactions:
await message.add_reaction(reaction)
s = db.Session()
s.add(db.Polls(message.id, ctx.channel.id, ctx.guild.id, ctx.author.id, reactions, multi))
s.commit()
s.close()
@commands.Cog.listener()
async def on_raw_reaction_add(self, payload: RawReactionActionEvent):

59
extensions/presentation.py
View file

@ -1,10 +1,13 @@
from discord.abc import GuildChannel
from discord.ext import commands
from discord import Embed, Message
from discord import Message, Role, TextChannel
from discord.ext.commands import BadArgument
from discord_slash import cog_ext, SlashCommandOptionType, SlashContext
from discord_slash.utils import manage_commands
from administrator.check import is_enabled
from administrator.check import is_enabled, guild_only, has_permissions
from administrator.logger import logger
from administrator import db
from administrator import db, slash
from administrator.utils import event_is_enabled
extension_name = "presentation"
@ -14,51 +17,49 @@ logger = logger.getChild(extension_name)
class Presentation(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
slash.get_cog_commands(self)
def description(self):
return "Give role to user who make a presentation in a dedicated channel"
@commands.group("presentation", pass_context=True)
@cog_ext.cog_subcommand(base="presentation", name="set",
description="Set the presentation channel and the role to give",
options=[
manage_commands.create_option("channel", "The presentation channel",
SlashCommandOptionType.CHANNEL, True),
manage_commands.create_option("role", "The role to give",
SlashCommandOptionType.ROLE, True)
])
@is_enabled()
@commands.guild_only()
@commands.has_permissions(manage_guild=True)
async def presentation(self, ctx: commands.Context):
if ctx.invoked_subcommand is None:
await ctx.invoke(self.presentation_help)
@presentation.group("help", pass_context=True)
async def presentation_help(self, ctx: commands.Context):
embed = Embed(title="Presentation help", description="Give a role to a new member after a presentation")
embed.add_field(name="set <#channel> <@role>", value="Set the presentation channel and the role to give",
inline=False)
embed.add_field(name="disable", value="Disable the auto role give", inline=False)
await ctx.send(embed=embed)
@presentation.group("set", pass_context=True)
async def presentation_set(self, ctx: commands.Context):
if len(ctx.message.channel_mentions) != 1 and not len(ctx.message.role_mentions) != 1:
@guild_only()
@has_permissions(manage_guild=True)
async def presentation_set(self, ctx: SlashContext, channel: GuildChannel, role: Role):
if not isinstance(channel, TextChannel):
raise BadArgument()
s = db.Session()
p = s.query(db.Presentation).filter(db.Presentation.guild == ctx.guild.id).first()
if not p:
p = db.Presentation(ctx.guild.id, ctx.message.channel_mentions[0].id, ctx.message.role_mentions[0].id)
p = db.Presentation(ctx.guild.id, channel.id, role.id)
s.add(p)
else:
p.channel = ctx.message.channel_mentions[0].id
p.role = ctx.message.role_mentions[0].id
p.channel = channel.id
p.role = role.id
s.commit()
await ctx.message.add_reaction("\U0001f44d")
await ctx.send(content="\U0001f44d")
@presentation.group("disable", pass_context=True)
async def presentation_disable(self, ctx: commands.Context):
@cog_ext.cog_subcommand(base="presentation", name="disable", description="Disable the auto role give")
@is_enabled()
@guild_only()
@has_permissions(manage_guild=True)
async def presentation_disable(self, ctx: SlashContext):
s = db.Session()
p = s.query(db.Presentation).filter(db.Presentation.guild == ctx.guild.id).first()
if not p:
await ctx.send(f"Nothing to disable !")
await ctx.send(content="Nothing to disable !")
else:
s.delete(p)
s.commit()
await ctx.message.add_reaction("\U0001f44d")
await ctx.send(content="\U0001f44d")
s.close()
@commands.Cog.listener()

40
extensions/purge.py
View file

@ -2,8 +2,10 @@ from asyncio import sleep
from discord.ext import commands
from discord import Embed, RawReactionActionEvent
from discord_slash import SlashContext, cog_ext
from administrator.check import is_enabled
from administrator import slash
from administrator.check import is_enabled, guild_only, has_permissions
from administrator.logger import logger
from administrator.utils import event_is_enabled
@ -15,34 +17,23 @@ class Purge(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
self.purges = {}
slash.get_cog_commands(self)
def description(self):
return "Purge all messages between the command and the next add reaction"
@commands.group("purge", pass_context=True)
@cog_ext.cog_slash(name="purge", description="Purge all message delimited by the command to your next reaction")
@is_enabled()
@commands.guild_only()
@commands.has_permissions(manage_messages=True)
async def purge(self, ctx: commands.Context):
if ctx.invoked_subcommand is None:
self.purges[ctx.message.author.id] = ctx.message
await ctx.message.add_reaction("\U0001f44d")
@guild_only()
@has_permissions(manage_messages=True)
async def purge(self, ctx: SlashContext):
message = await ctx.channel.send(content="\U0001f44d")
self.purges[ctx.author.id] = message
await sleep(2*60)
try:
if self.purges[ctx.message.author.id] == ctx.message:
await ctx.message.clear_reactions()
del self.purges[ctx.message.author.id]
except:
pass
@purge.group("help", pass_context=True)
@commands.guild_only()
async def purge_help(self, ctx: commands.Context):
embed = Embed(title="Purge help")
embed.add_field(name="purge", value="Purge all message delimited by the command to your next reaction",
inline=False)
await ctx.send(embed=embed)
await sleep(2*60)
if ctx.author.id in self.purges and self.purges[ctx.author.id] == message:
await message.delete()
del self.purges[ctx.author.id]
@commands.Cog.listener()
async def on_raw_reaction_add(self, payload: RawReactionActionEvent):
@ -55,8 +46,7 @@ class Purge(commands.Cog):
if user.id in self.purges:
if message.channel == self.purges[user.id].channel:
async with message.channel.typing():
await message.channel.purge(before=self.purges[user.id], after=message,
limit=None)
await message.channel.purge(before=self.purges[user.id], after=message, limit=None)
await self.purges[user.id].delete()
await message.delete()
del self.purges[user.id]

70
extensions/reminder.py
View file

@ -1,14 +1,15 @@
import re
from datetime import datetime, timedelta
from datetime import datetime
from discord.ext import commands
from discord import Embed
from discord.ext.commands import BadArgument
from discord.ext import tasks
from discord_slash import SlashContext, cog_ext, SlashCommandOptionType
from discord_slash.utils import manage_commands
from administrator.check import is_enabled
from administrator.logger import logger
from administrator import db
from administrator import db, slash
from administrator.utils import time_pars, seconds_to_time_string
extension_name = "reminders"
@ -18,61 +19,52 @@ logger = logger.getChild(extension_name)
class Reminders(commands.Cog, name="Reminder"):
def __init__(self, bot: commands.Bot):
self.bot = bot
slash.get_cog_commands(self)
def description(self):
return "Create and manage reminders"
@commands.group("reminder", pass_context=True)
@cog_ext.cog_subcommand(base="reminder", name="add", description="Add a reminder to your reminders list", options=[
manage_commands.create_option("message", "The message", SlashCommandOptionType.STRING, True),
manage_commands.create_option("time", "When, ?D?H?M?S", SlashCommandOptionType.STRING, True)
])
@is_enabled()
async def reminder(self, ctx: commands.Context):
if ctx.invoked_subcommand is None:
await ctx.invoke(self.reminder_help)
@reminder.group("help", pass_context=True)
async def reminder_help(self, ctx: commands.Context):
embed = Embed(title="Reminder help")
embed.add_field(name="reminder add <message> <time>", value="Add a reminder to your reminders list\n"
"Time: ?D?H?M?S", inline=False)
embed.add_field(name="reminder list", value="Show your tasks list", inline=False)
embed.add_field(name="reminder remove [N°]", value="Show your tasks list with if no id given\n"
"Remove the task withe the matching id", inline=False)
await ctx.send(embed=embed)
@reminder.group("add", pass_context=True)
async def reminder_add(self, ctx: commands.Context, message: str, time: str):
async def reminder_add(self, ctx: SlashContext, message: str, time: str):
time = time_pars(time)
now = datetime.now()
s = db.Session()
s.add(db.Task(message, ctx.author.id, ctx.channel.id, now + time, ctx.message.created_at))
s.add(db.Task(message, ctx.author.id, ctx.channel.id, now + time, datetime.now()))
s.commit()
s.close()
await ctx.send(f"""Remind you in {seconds_to_time_string(time.total_seconds())} !""")
await ctx.send(content=f"""Remind you in {seconds_to_time_string(time.total_seconds())} !""")
@reminder.group("list", pass_context=True)
async def reminder_list(self, ctx: commands.Context):
@cog_ext.cog_subcommand(base="reminder", name="list", description="Show your tasks list")
@is_enabled()
async def reminder_list(self, ctx: SlashContext):
embed = Embed(title="Tasks list")
s = db.Session()
for t in s.query(db.Task).filter(db.Task.user == ctx.author.id).all():
embed.add_field(name=f"N°{t.id} | {t.date.strftime('%d/%m/%Y %H:%M')}", value=f"{t.message}", inline=False)
s.close()
await ctx.send(embed=embed)
await ctx.send(embeds=[embed])
@reminder.group("remove", pass_context=True)
async def reminder_remove(self, ctx: commands.Context, n: int = None):
if n is None:
await ctx.invoke(self.reminder_list)
@cog_ext.cog_subcommand(base="reminder", name="remove", description="Remove the task withe the matching id",
options=[
manage_commands.create_option("id", "The reminder id",
SlashCommandOptionType.INTEGER, True)])
@is_enabled()
async def reminder_remove(self, ctx: SlashContext, n: int):
s = db.Session()
t = s.query(db.Task).filter(db.Task.id == n).first()
if t and t.user == ctx.author.id:
s.delete(t)
s.commit()
s.close()
await ctx.send(content="\U0001f44d")
else:
s = db.Session()
t = s.query(db.Task).filter(db.Task.id == n).first()
if t and t.user == ctx.author.id:
s.delete(t)
s.commit()
s.close()
await ctx.message.add_reaction("\U0001f44d")
else:
s.close()
raise BadArgument()
s.close()
raise BadArgument()
@tasks.loop(minutes=1)
async def reminders_loop(self):

187
extensions/rorec.py
View file

@ -1,84 +1,64 @@
import re
from discord.abc import GuildChannel
from discord.ext import commands
from discord import Embed, RawReactionActionEvent, RawBulkMessageDeleteEvent, RawMessageDeleteEvent, NotFound, \
InvalidArgument, HTTPException, TextChannel, Forbidden
InvalidArgument, HTTPException, TextChannel, Forbidden, Role, Message
from discord.ext.commands import BadArgument
from discord_slash import cog_ext, SlashContext, SlashCommandOptionType
from discord_slash.utils import manage_commands
from administrator import db
from administrator.check import is_enabled
from administrator import db, slash
from administrator.check import is_enabled, guild_only, has_permissions
from administrator.logger import logger
from administrator.utils import event_is_enabled
from administrator.utils import event_is_enabled, get_message_by_url
extension_name = "rorec"
logger = logger.getChild(extension_name)
channel_id_re = re.compile(r"^<#([0-9]+)>$")
class RoRec(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
slash.get_cog_commands(self)
def description(self):
return "Create role-reaction message to give role from a reaction add"
@staticmethod
def get_message(session: db.Session, message_id: int, guild_id: int) -> db.RoRec:
m = session.query(db.RoRec).filter(db.RoRec.message == message_id and db.RoRec.guild == guild_id).first()
async def get_message(session: db.Session, ctx: SlashContext, url: str) -> db.RoRec:
m = session.query(db.RoRec).filter(db.RoRec.message == (await get_message_by_url(ctx, url)).id and
db.RoRec.guild == ctx.guild.id).first()
if not m:
raise BadArgument()
else:
return m
async def try_emoji(self, ctx: commands.Context, emoji: str):
async def try_emoji(self, msg: Message, emoji: str):
try:
await ctx.message.add_reaction(emoji)
await msg.add_reaction(emoji)
except (HTTPException, NotFound, InvalidArgument):
raise BadArgument()
else:
await (await ctx.channel.fetch_message(ctx.message.id)).remove_reaction(emoji, self.bot.user)
await (await msg.channel.fetch_message(msg.id)).remove_reaction(emoji, self.bot.user)
@commands.group("rorec", pass_context=True)
@cog_ext.cog_subcommand(base="rorec", name="new",
description="Create a new role-reaction message on the mentioned channel",
options=[
manage_commands.create_option("title", "The title",
SlashCommandOptionType.STRING, True),
manage_commands.create_option("channel", "The target channel",
SlashCommandOptionType.CHANNEL, True),
manage_commands.create_option("description", "The description",
SlashCommandOptionType.STRING, False),
manage_commands.create_option("one", "If only one role is packable",
SlashCommandOptionType.BOOLEAN, False)
])
@is_enabled()
@commands.guild_only()
@commands.has_permissions(manage_roles=True)
async def rorec(self, ctx: commands.Context):
if ctx.invoked_subcommand is None:
await ctx.invoke(self.rorec_help)
@rorec.group("help", pass_context=True)
async def rorec_help(self, ctx: commands.Context):
embed = Embed(title="Role-Reaction help")
embed.add_field(name="new <title> <#channel> [description] [Only one (True/False)]",
value="Create a new role-reaction message on the mentioned channel.\n"
"You can specify a description and if you can pick only one role",
inline=False)
embed.add_field(name="edit <message id> <title> [description}", value="Edit a role-reaction message title and "
"description",
inline=False)
embed.add_field(name="set <message_id> <emoji> <@role1> [@role2] ...",
value="Add/edit a emoji with linked roles", inline=False)
embed.add_field(name="remove <message_id> <emoji>", value="Remove a emoji of a role-reaction message",
inline=False)
embed.add_field(name="reload <message_id>", value="Reload the message and the reactions", inline=False)
embed.add_field(name="delete <message_id>", value="Remove a role-reaction message", inline=False)
await ctx.send(embed=embed)
@rorec.group("new", pass_context=True)
async def rorec_new(self, ctx: commands.Context, title: str, channel: str, description: str = "",
@guild_only()
@has_permissions(manage_roles=True)
async def rorec_new(self, ctx: SlashContext, title: str, channel: GuildChannel, description: str = "",
one: bool = False):
channel = channel_id_re.findall(channel)
if len(channel) != 1:
if not isinstance(channel, TextChannel):
raise BadArgument()
channel = ctx.guild.get_channel(int(channel[0]))
if not channel:
raise BadArgument()
if description in ["True", "False"]:
one = True if description == "True" else False
description = ""
embed = Embed(title=title, description=description)
embed.add_field(name="Roles", value="No role yet...")
@ -88,12 +68,24 @@ class RoRec(commands.Cog):
s.add(r)
s.commit()
s.close()
await ctx.message.add_reaction("\U0001f44d")
await ctx.send(content="\U0001f44d")
@rorec.group("edit", pass_context=True)
async def rorec_edit(self, ctx: commands.Context, message_id: int, title: str, description: str = None):
@cog_ext.cog_subcommand(base="rorec", name="edit",
description="Edit a role-reaction message title and description",
options=[
manage_commands.create_option("url", "The message url",
SlashCommandOptionType.STRING, True),
manage_commands.create_option("title", "The new title",
SlashCommandOptionType.STRING, True),
manage_commands.create_option("description", "The new description",
SlashCommandOptionType.STRING, False)
])
@is_enabled()
@guild_only()
@has_permissions(manage_roles=True)
async def rorec_edit(self, ctx: SlashContext, url: str, title: str, description: str = ""):
s = db.Session()
m = self.get_message(s, message_id, ctx.guild.id)
m = await self.get_message(s, ctx, url)
s.close()
message = await ctx.guild.get_channel(m.channel).fetch_message(m.message)
@ -101,31 +93,57 @@ class RoRec(commands.Cog):
embed.title = title
embed.description = description
await message.edit(embed=embed)
await ctx.message.add_reaction("\U0001f44d")
await ctx.send(content="\U0001f44d")
@rorec.group("set", pass_context=True)
async def rorec_set(self, ctx: commands.Context, message_id: int, emoji: str):
@cog_ext.cog_subcommand(base="rorec", name="set",
description="Add/edit a emoji with linked roles",
options=[
manage_commands.create_option("url", "The message url",
SlashCommandOptionType.STRING, True),
manage_commands.create_option("emoji", "The emoji",
SlashCommandOptionType.STRING, True),
manage_commands.create_option("role", "The role",
SlashCommandOptionType.ROLE, True)
])
@is_enabled()
@guild_only()
@has_permissions(manage_roles=True)
async def rorec_set(self, ctx: SlashContext, url: str, emoji: str, role: Role):
await ctx.send(content="\U000023f3")
s = db.Session()
m = self.get_message(s, message_id, ctx.guild.id)
m = await self.get_message(s, ctx, url)
if len(ctx.message.role_mentions) == 0:
raise BadArgument()
await self.try_emoji(ctx, emoji)
await ctx.delete()
msg = await ctx.channel.send("\U000023f3")
await self.try_emoji(msg, emoji)
data = m.get_data()
data[emoji] = list(map(lambda x: x.id, ctx.message.role_mentions))
data[emoji] = list(map(lambda x: x.id, [role]))
m.set_data(data)
await self.rorec_update(m)
s.commit()
s.close()
await ctx.message.add_reaction("\U0001f44d")
await msg.edit(content="\U0001f44d")
@rorec.group("remove", pass_context=True)
async def rorec_remove(self, ctx: commands.Context, message_id: int, emoji: str):
@cog_ext.cog_subcommand(base="rorec", name="remove",
description="Remove a emoji of a role-reaction message",
options=[
manage_commands.create_option("url", "The message url",
SlashCommandOptionType.STRING, True),
manage_commands.create_option("emoji", "The emoji",
SlashCommandOptionType.STRING, True)
])
@is_enabled()
@guild_only()
@has_permissions(manage_roles=True)
async def rorec_remove(self, ctx: SlashContext, url: str, emoji: str):
await ctx.send(content="\U000023f3")
s = db.Session()
m = self.get_message(s, message_id, ctx.guild.id)
await self.try_emoji(ctx, emoji)
m = await self.get_message(s, ctx, url)
await ctx.delete()
msg = await ctx.channel.send("\U000023f3")
await self.try_emoji(msg, emoji)
data = m.get_data()
if emoji not in data:
@ -136,24 +154,37 @@ class RoRec(commands.Cog):
await self.rorec_update(m)
s.commit()
s.close()
await ctx.message.add_reaction("\U0001f44d")
await msg.edit("\U0001f44d")
@rorec.group("reload", pass_context=True)
async def rorec_reload(self, ctx: commands.Context, message_id: int):
@cog_ext.cog_subcommand(base="rorec", name="reload",
description="Reload the message and the reactions",
options=[manage_commands.create_option("url", "The message url",
SlashCommandOptionType.STRING, True)])
@is_enabled()
@guild_only()
@has_permissions(manage_roles=True)
async def rorec_reload(self, ctx: SlashContext, url: str):
s = db.Session()
m = self.get_message(s, message_id, ctx.guild.id)
m = await self.get_message(s, ctx, url)
await self.rorec_update(m)
s.close()
await ctx.message.add_reaction("\U0001f44d")
await ctx.send(content="\U0001f44d")
@rorec.group("delete", pass_context=True)
async def rorec_delete(self, ctx: commands.Context, message_id: int):
@cog_ext.cog_subcommand(base="rorec", name="delete",
description="Remove a role-reaction message",
options=[manage_commands.create_option("url", "The message link",
SlashCommandOptionType.STRING, True)])
@is_enabled()
@guild_only()
@has_permissions(manage_roles=True)
async def rorec_delete(self, ctx: SlashContext, url: str):
msg = await get_message_by_url(ctx, url)
s = db.Session()
m = self.get_message(s, message_id, ctx.guild.id)
await self.get_message(s, ctx, url)
s.close()
await (await self.bot.get_channel(m.channel).fetch_message(m.message)).delete()
await ctx.message.add_reaction("\U0001f44d")
await msg.delete()
await ctx.send(content="\U0001f44d")
async def rorec_update(self, m: db.RoRec):
channel = self.bot.get_channel(m.channel)
@ -173,6 +204,8 @@ class RoRec(commands.Cog):
value += ", ".join(map(lambda x: self.bot.get_guild(m.guild).get_role(x).mention, data[d]))
value += "\n"
await message.add_reaction(d)
if not value:
value = "No role yet..."
embed.add_field(name=name, value=value)
await message.edit(embed=embed)

64
extensions/speak.py
View file

@ -1,8 +1,11 @@
from discord.ext import commands
from discord import Member, VoiceState, Embed, Reaction, Guild
from discord.ext.commands import CommandNotFound
from discord.ext.commands import BadArgument
from discord_slash import SlashContext, cog_ext, SlashCommandOptionType
from discord_slash.utils import manage_commands
from administrator.check import is_enabled
from administrator import slash
from administrator.check import is_enabled, guild_only, has_permissions
from administrator.logger import logger
from administrator.utils import event_is_enabled
@ -21,29 +24,24 @@ class Speak(commands.Cog):
self.last_reaction = None
self.voice_message = None
self.last_message = None
slash.get_cog_commands(self)
def description(self):
return "Speech manager"
@commands.group("speak", pass_context=True)
@cog_ext.cog_subcommand(base="speak", name="setup",
description="Set your current voice channel as the speak channel",
options=[
manage_commands.create_option("strict", "Mute everyone on setup",
SlashCommandOptionType.BOOLEAN, False)
])
@is_enabled()
@commands.guild_only()
@commands.has_guild_permissions(mute_members=True)
async def speak(self, ctx: commands.Context):
if ctx.invoked_subcommand is None:
raise CommandNotFound
@speak.group("help", pass_context=True)
async def speak_help(self, ctx: commands.Context):
embed = Embed(title="Speak help")
embed.add_field(name="speak setup [strict]",
value="Set your current voice channel as the speak channel", inline=False)
embed.add_field(name="speak mute", value="Mute everyone on the speak channel except you", inline=False)
embed.add_field(name="speak unmute", value="Unmute everyone on the speak channel except you", inline=False)
await ctx.send(embed=embed)
@speak.group("setup", pass_context=True)
async def speak_setup(self, ctx: commands.Context, *args):
@guild_only()
@has_permissions(mute_members=True)
async def speak_setup(self, ctx: SlashContext, strict: bool = False):
self.strict = strict
if not ctx.author.voice:
raise BadArgument()
self.voice_chan = ctx.author.voice.channel.id
embed = Embed(title="Speak \U0001f508")
embed.add_field(name="Waiting list \u23f3", value="Nobody", inline=False)
@ -57,24 +55,30 @@ class Speak(commands.Cog):
"\u274C Clear the speak\n"
"Remove your reaction to remove from list",
inline=False)
self.voice_message = await ctx.send(embed=embed)
self.voice_message = await ctx.channel.send(embed=embed)
for reaction in ["\U0001f5e3", "\u2757", "\u27A1", "\U0001F512", "\U0001F507", "\U0001F50A", "\u274C"]:
await self.voice_message.add_reaction(reaction)
self.voice_message = await self.voice_message.channel.fetch_message(self.voice_message.id)
@speak.group("mute", pass_context=True)
async def speak_mute(self, ctx: commands.Context):
@cog_ext.cog_subcommand(base="speak", name="mute", description="Mute everyone on the speak channel except you")
@is_enabled()
@guild_only()
@has_permissions(mute_members=True)
async def speak_mute(self, ctx: SlashContext):
if not await self.mute(True, ctx.author):
await ctx.message.add_reaction("\u274C")
await ctx.send(content="\u274C")
else:
await ctx.message.add_reaction("\U0001f44d")
await ctx.send(content="\U0001f44d")
@speak.group("unmute", pass_context=True)
async def speak_unmute(self, ctx: commands.Context):
@cog_ext.cog_subcommand(base="speak", name="unmute", description="Unmute everyone on the speak channel except you")
@is_enabled()
@guild_only()
@has_permissions(mute_members=True)
async def speak_unmute(self, ctx: SlashContext):
if not await self.mute(False, ctx.author):
await ctx.message.add_reaction("\u274C")
await ctx.send(content="\u274C")
else:
await ctx.message.add_reaction("\U0001f44d")
await ctx.send(content="\U0001f44d")
@commands.Cog.listener()
async def on_voice_state_update(self, member: Member, before: VoiceState, after: VoiceState):
@ -91,7 +95,7 @@ class Speak(commands.Cog):
(after is not None and after.channel is not None and after.channel.id != self.voice_chan):
await member.edit(mute=False)
async def cog_after_invoke(self, ctx: commands.Context):
async def cog_after_invoke(self, ctx: SlashContext):
await ctx.message.delete(delay=30)
@commands.Cog.listener()

24
extensions/tex.py
View file

@ -1,8 +1,10 @@
from urllib.parse import urlencode
from discord import Embed
from discord.ext import commands
from discord_slash import cog_ext, SlashContext, SlashCommandOptionType
from discord_slash.utils import manage_commands
from administrator import slash
from administrator.check import is_enabled
from administrator.logger import logger
@ -14,26 +16,16 @@ class TeX(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
self.polls = {}
slash.get_cog_commands(self)
def description(self):
return "Render TeX formula"
@commands.group("tex", pass_context=True)
@cog_ext.cog_slash(name="tex", description="Render a TeX formula", options=[
manage_commands.create_option("formula", "The TeX formula", SlashCommandOptionType.STRING, True)])
@is_enabled()
async def tex(self, ctx: commands.Context):
if ctx.message.content.count("`") == 2:
start = ctx.message.content.find("`")
end = ctx.message.content.find("`", start+1)
command = ctx.message.content[start+1:end]
await ctx.send(f"https://chart.apis.google.com/chart?cht=tx&chs=40&{urlencode({'chl': command})}")
elif ctx.invoked_subcommand is None:
await ctx.invoke(self.tex_help)
@tex.group("help", pass_context=True)
async def tex_help(self, ctx: commands.Context):
embed = Embed(title="TeX help")
embed.add_field(name="tex \`formula\`", value="Render a TeX formula", inline=False)
await ctx.send(embed=embed)
async def tex(self, ctx: SlashContext, formula: str):
await ctx.send(content=f"https://chart.apis.google.com/chart?cht=tx&chs=40&{urlencode({'chl': formula})}")
def setup(bot):

30
extensions/tomuss.py
View file

@ -5,9 +5,12 @@ from time import mktime
from discord import Embed, Forbidden, HTTPException
from discord.ext import commands, tasks
from discord.ext.commands import BadArgument
from discord_slash import SlashContext, cog_ext, SlashCommandOptionType
from discord_slash.utils import manage_commands
from feedparser import parse
import db
from administrator import slash
from administrator.check import is_enabled
from administrator.logger import logger
@ -21,25 +24,14 @@ class Tomuss(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
self.tomuss_loop.start()
slash.get_cog_commands(self)
def description(self):
return "PCP Univ Lyon 1"
@commands.group("tomuss", pass_context=True)
@is_enabled()
async def tomuss(self, ctx: commands.Context):
if ctx.invoked_subcommand is None:
await ctx.invoke(self.tomuss_help)
@tomuss.group("help", pass_context=True)
async def tomuss_help(self, ctx: commands.Context):
embed = Embed(title="Tomuss help")
embed.add_field(name="tomuss set <url>", value="Set your tomuss RSS feed", inline=False)
embed.add_field(name="tomuss unset", value="Unset your tomuss RSS feed", inline=False)
await ctx.send(embed=embed)
@tomuss.group("set", pass_context=True)
async def tomuss_set(self, ctx: commands.Context, url: str):
@cog_ext.cog_subcommand(base="tomuss", name="set", description="Set your tomuss RSS feed", options=[
manage_commands.create_option("url", "The RSS URL", SlashCommandOptionType.STRING, True)])
async def tomuss_set(self, ctx: SlashContext, url: str):
if not url_re.fullmatch(url):
raise BadArgument()
entries = parse(url).entries
@ -59,10 +51,10 @@ class Tomuss(commands.Cog):
s.commit()
s.close()
await ctx.message.add_reaction("\U0001f44d")
await ctx.channel.send(f"Tomuss RSS set for {ctx.author.mention} \U0001f44d")
@tomuss.group("unset", pass_context=True)
async def tomuss_unset(self, ctx: commands.Context):
@cog_ext.cog_subcommand(base="tomuss", name="unset", description="Unset your tomuss RSS feed")
async def tomuss_unset(self, ctx: SlashContext):
s = db.Session()
t = s.query(db.Tomuss).get(ctx.author.id)
if not t:
@ -70,7 +62,7 @@ class Tomuss(commands.Cog):
s.delete(t)
s.commit()
s.close()
await ctx.message.add_reaction("\U0001f44d")
await ctx.send(content="\U0001f44d")
@tasks.loop(minutes=5)
async def tomuss_loop(self):

67
extensions/utils.py
View file

@ -1,9 +1,11 @@
from datetime import datetime
from discord import Embed, Member, Guild
from discord import Embed, Guild
from discord.ext import commands
from discord.ext.commands import BadArgument, CommandError
from discord_slash import cog_ext, SlashContext, SlashCommandOptionType
from discord_slash.utils import manage_commands
from administrator import slash
from administrator.check import is_enabled
from administrator.logger import logger
@ -15,34 +17,17 @@ logger = logger.getChild(extension_name)
class Utils(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
slash.get_cog_commands(self)
def description(self):
return "Some tools"
@commands.group("utils", pass_context=True)
@is_enabled()
async def utils(self, ctx: commands.Context):
if ctx.invoked_subcommand is None:
await ctx.invoke(self.utils_help)
@utils.group("help", pass_context=True)
async def utils_help(self, ctx: commands.Context):
embed = Embed(title="Utils help")
try:
if await self.eval.can_run(ctx):
embed.add_field(name="eval \`\`\`code\`\`\`", value="Execute some code", inline=False)
except CommandError:
pass
embed.add_field(name="ping", value="Return the ping with the discord API", inline=False)
embed.add_field(name="info [@user]", value="Show information on guild or user specified", inline=False)
await ctx.send(embed=embed)
@commands.group("eval", pass_context=True)
@commands.is_owner()
async def eval(self, ctx: commands.Context):
start = ctx.message.content.find("```")
end = ctx.message.content.find("```", start+3)
command = ctx.message.content[start+3:end]
start = ctx.message.content.find("```python")
end = ctx.message.content.find("```", start+9)
command = ctx.message.content[start+9:end]
try:
exec("async def __ex(self, ctx):\n" + command.replace("\n", "\n "))
out = str(await locals()["__ex"](self, ctx))
@ -55,20 +40,20 @@ class Utils(commands.Cog):
except Exception as e:
await ctx.send(f"```{e.__class__.__name__}: {e}```")
@commands.group("ping", pass_context=True)
@cog_ext.cog_slash(name="ping", description="Return the ping with the discord API")
@is_enabled()
async def ping(self, ctx: commands.Context):
async def ping(self, ctx: SlashContext):
start = datetime.now()
msg = await ctx.send(f"Discord WebSocket latency: `{round(self.bot.latency*1000)}ms`")
await msg.edit(content=msg.content+"\n"+f"Bot latency: `{round((msg.created_at - start).microseconds/1000)}ms`")
content = f"Discord WebSocket latency: `{round(self.bot.latency*1000)}ms`"
await ctx.send(content=content)
await ctx.edit(content=content+"\n"+f"Bot latency: `{round((datetime.now() - start).microseconds/1000)}ms`")
@commands.group("info", pass_context=True)
@cog_ext.cog_slash(name="info",
description="Show information on guild or user specified",
options=[manage_commands.create_option("user", "A user", SlashCommandOptionType.USER, False)])
@is_enabled()
async def info(self, ctx: commands.Context):
if len(ctx.message.mentions) > 1:
raise BadArgument()
elif ctx.message.mentions:
user: Member = ctx.message.mentions[0]
async def info(self, ctx: SlashContext, user: SlashCommandOptionType.USER = None):
if user:
embed = Embed(title=str(user))
embed.set_author(name="User infos", icon_url=user.avatar_url)
embed.add_field(name="Display name", value=user.display_name)
@ -129,11 +114,11 @@ class Utils(commands.Cog):
embed.add_field(name="Shard ID", value=guild.shard_id)
embed.add_field(name="Created at", value=guild.created_at)
await ctx.send(embed=embed)
await ctx.send(embeds=[embed])
@commands.group("about", pass_context=True)
@cog_ext.cog_slash(name="about", description="Show information about the bot")
@is_enabled()
async def about(self, ctx: commands.Context):
async def about(self, ctx: SlashContext):
embed = Embed(title=self.bot.user.display_name, description=self.bot.description)
embed.set_author(name="Administrator", icon_url=self.bot.user.avatar_url, url="https://github.com/flifloo")
flifloo = self.bot.get_user(177393521051959306)
@ -142,9 +127,12 @@ class Utils(commands.Cog):
value=(await self.bot.application_info()).owner.display_name)
embed.add_field(name="Guilds", value=str(len(self.bot.guilds)))
embed.add_field(name="Extensions", value=str(len(self.bot.extensions)))
embed.add_field(name="Commands", value=str(len(self.bot.all_commands)))
embed.add_field(name="Commands", value=str(len(self.bot.all_commands)+len(slash.commands)))
embed.add_field(name="Latency", value=f"{round(self.bot.latency*1000)} ms")
await ctx.send(embed=embed)
await ctx.send(embeds=[embed])
def cog_unload(self):
slash.remove_cog_commands(self)
def setup(bot):
@ -165,3 +153,6 @@ def teardown(bot):
logger.error(f"Error unloading: {e}")
else:
logger.info(f"Unload successful")

166
extensions/warn.py
View file

@ -1,9 +1,11 @@
from discord import Embed, Forbidden, Member, Guild
from discord.ext import commands
from discord.ext.commands import BadArgument
from discord_slash import cog_ext, SlashCommandOptionType, SlashContext
from discord_slash.utils import manage_commands
from administrator import db
from administrator.check import is_enabled
from administrator import db, slash
from administrator.check import is_enabled, guild_only, has_permissions
from administrator.logger import logger
from administrator.utils import time_pars, seconds_to_time_string
@ -14,12 +16,13 @@ logger = logger.getChild(extension_name)
class Warn(commands.Cog):
def __init__(self, bot: commands.Bot):
self.bot = bot
slash.get_cog_commands(self)
def description(self):
return "Send warning to user and make custom action after a number of warn"
@staticmethod
async def check_warn(ctx: commands.Context, target: Member):
async def check_warn(ctx: SlashContext, target: Member):
s = db.Session()
c = s.query(db.Warn).filter(db.Warn.guild == ctx.guild.id, db.Warn.user == target.id).count()
a = s.query(db.WarnAction).filter(db.WarnAction.guild == ctx.guild.id, db.WarnAction.count == c).first()
@ -32,96 +35,72 @@ class Warn(commands.Cog):
elif a.action == "mute":
pass # Integration with upcoming ban & mute extension
@staticmethod
def get_target(ctx: commands.Context, user: str) -> Member:
users = {str(m): m for m in ctx.guild.members}
if user not in users:
raise BadArgument()
return users[user]
@commands.group("warn", pass_context=True)
@cog_ext.cog_subcommand(base="warn", name="add", description="Send a warn to a user", options=[
manage_commands.create_option("user", "The user", SlashCommandOptionType.USER, True),
manage_commands.create_option("description", "The description", SlashCommandOptionType.STRING, True)
])
@is_enabled()
@commands.guild_only()
#@commands.has_permissions(manage_roles=True, kick_members=True, ban_members=True, mute_members=True)
async def warn(self, ctx: commands.Context):
if ctx.invoked_subcommand is None:
await ctx.invoke(self.warn_help)
@warn.group("help", pass_context=True)
async def warn_help(self, ctx: commands.Context):
embed = Embed(title="Warn help")
embed.add_field(name="add <user> <description>", value="Send a warn to a user", inline=False)
embed.add_field(name="remove <user> <number>", value="Remove a number of warn to a user", inline=False)
embed.add_field(name="purge <user>", value="Remove all warn of a user", inline=False)
embed.add_field(name="list [user#discriminator|actions]", value="List warn of the guild or a specified user\n"
"If you specify `actions` instead of a user, "
"all the actions of the guild will be listed",
inline=False)
embed.add_field(name="action <count> <action>", value="Set an action for a count of warn\n"
"Actions: `mute<time>`, `kick`, `ban[time]`, `nothing`\n"
"Time: `?D?H?M?S`\n"
"Example: `action 1 mute1H` to mute someone for one hour "
"after only one war\n"
"or `action 3 ban1D` to ban someone for one day after 3 "
"warns", inline=False)
await ctx.send(embed=embed)
@warn.group("add", pass_context=True)
async def warn_add(self, ctx: commands.Context, user: str, description: str):
target = self.get_target(ctx, user)
@guild_only()
@has_permissions(kick_members=True, ban_members=True, mute_members=True)
async def warn_add(self, ctx: SlashContext, user: Member, description: str):
s = db.Session()
s.add(db.Warn(target.id, ctx.author.id, ctx.guild.id, description))
s.add(db.Warn(user.id, ctx.author.id, ctx.guild.id, description))
s.commit()
s.close()
try:
embed = Embed(title="You get warned !", description="A moderator send you a warn", color=0xff0000)
embed.add_field(name="Description:", value=description)
await target.send(embed=embed)
await user.send(embed=embed)
except Forbidden:
await ctx.send("Fail to send warn notification to the user, DM close :warning:")
await ctx.send(content="Fail to send warn notification to the user, DM close :warning:")
else:
await ctx.message.add_reaction("\U0001f44d")
await self.check_warn(ctx, target)
await ctx.send(content="\U0001f44d")
await self.check_warn(ctx, user)
@warn.group("remove", pass_context=True)
async def warn_remove(self, ctx: commands.Context, user: str, number: int):
target = self.get_target(ctx, user)
@cog_ext.cog_subcommand(base="warn", name="remove", description="Remove a number of warn to a user", options=[
manage_commands.create_option("user", "The user", SlashCommandOptionType.USER, True),
manage_commands.create_option("number", "The warn to remove", SlashCommandOptionType.INTEGER, True)
])
@is_enabled()
@guild_only()
@has_permissions(kick_members=True, ban_members=True, mute_members=True)
async def warn_remove(self, ctx: SlashContext, user: Member, number: int):
s = db.Session()
ws = s.query(db.Warn).filter(db.Warn.guild == ctx.guild.id, db.Warn.user == target.id).all()
ws = s.query(db.Warn).filter(db.Warn.guild == ctx.guild.id, db.Warn.user == user.id).all()
if number <= 0 or number > len(ws):
raise BadArgument()
s.delete(ws[number-1])
s.commit()
s.close()
await ctx.message.add_reaction("\U0001f44d")
await ctx.send(content="\U0001f44d")
@warn.group("purge", pass_context=True)
async def warn_purge(self, ctx: commands.Context, user: str):
target = self.get_target(ctx, user)
@cog_ext.cog_subcommand(base="warn", name="purge", description="Remove all warn of a user", options=[
manage_commands.create_option("user", "The user", SlashCommandOptionType.USER, True)])
@is_enabled()
@guild_only()
@has_permissions(kick_members=True, ban_members=True, mute_members=True)
async def warn_purge(self, ctx: SlashContext, user: Member):
s = db.Session()
for w in s.query(db.Warn).filter(db.Warn.guild == ctx.guild.id, db.Warn.user == target.id).all():
for w in s.query(db.Warn).filter(db.Warn.guild == ctx.guild.id, db.Warn.user == user.id).all():
s.delete(w)
s.commit()
s.close()
await ctx.message.add_reaction("\U0001f44d")
await ctx.send(content="\U0001f44d")
@warn.group("list", pass_context=True)
async def warn_list(self, ctx: commands.Context, user: str = None):
@cog_ext.cog_subcommand(base="warn", name="list", description="List warn of the guild or a specified user",
options=[
manage_commands.create_option("user", "The user", SlashCommandOptionType.USER, False)
])
@is_enabled()
@guild_only()
@has_permissions(kick_members=True, ban_members=True, mute_members=True)
async def warn_list(self, ctx: SlashContext, user: Member = None):
s = db.Session()
embed = Embed(title="Warn list")
ws = {}
if user == "actions":
embed.title = "Actions list"
for a in s.query(db.WarnAction).filter(db.WarnAction.guild == ctx.guild.id).order_by(db.WarnAction.count)\
.all():
action = f"{a.action} for {seconds_to_time_string(a.duration)}" if a.duration else a.action
embed.add_field(name=f"{a.count} warn(s)", value=action, inline=False)
elif user:
target = self.get_target(ctx, user)
ws[target.id] = s.query(db.Warn).filter(db.Warn.guild == ctx.guild.id, db.Warn.user == target.id).all()
if user:
ws[user.id] = s.query(db.Warn).filter(db.Warn.guild == ctx.guild.id, db.Warn.user == user.id).all()
else:
for w in s.query(db.Warn).filter(db.Warn.guild == ctx.guild.id).all():
if w.user not in ws:
@ -134,10 +113,45 @@ class Warn(commands.Cog):
for w in ws[u]]
embed.add_field(name=self.bot.get_user(u), value="\n".join(warns), inline=False)
await ctx.send(embed=embed)
await ctx.send(embeds=[embed])
@warn.group("action", pass_context=True)
async def warn_action(self, ctx: commands.Context, count: int, action: str):
@cog_ext.cog_subcommand(base="warn", name="actions", description="List all the actions of the guild")
@is_enabled()
@guild_only()
@has_permissions(kick_members=True, ban_members=True, mute_members=True)
async def warn_actions(self, ctx: SlashContext):
s = db.Session()
embed = Embed(title="Warn list")
ws = {}
embed.title = "Actions list"
for a in s.query(db.WarnAction).filter(db.WarnAction.guild == ctx.guild.id).order_by(db.WarnAction.count) \
.all():
action = f"{a.action} for {seconds_to_time_string(a.duration)}" if a.duration else a.action
embed.add_field(name=f"{a.count} warn(s)", value=action, inline=False)
s.close()
for u in ws:
warns = [f"{self.bot.get_user(w.author).mention} - {w.date.strftime('%d/%m/%Y %H:%M')}```{w.description}```"
for w in ws[u]]
embed.add_field(name=self.bot.get_user(u), value="\n".join(warns), inline=False)
await ctx.send(embeds=[embed])
@cog_ext.cog_subcommand(base="warn", name="action", description="Set an action for a count of warn", options=[
manage_commands.create_option("count", "The number of warns", SlashCommandOptionType.INTEGER, True),
manage_commands.create_option("action", "The action", SlashCommandOptionType.STRING, True, [
manage_commands.create_choice("mute", "mute"),
manage_commands.create_choice("kick", "kick"),
manage_commands.create_choice("ban", "ban"),
manage_commands.create_choice("nothing", "nothing")
]),
manage_commands.create_option("time", "The duration of the action, ?D?H?M?S",
SlashCommandOptionType.STRING, False)
])
@is_enabled()
@guild_only()
@has_permissions(administrator=True)
async def warn_action(self, ctx: SlashContext, count: int, action: str, time: str = None):
if count <= 0 or\
(action not in ["kick", "nothing"] and not action.startswith("mute") and not action.startswith("ban")):
raise BadArgument()
@ -151,14 +165,8 @@ class Warn(commands.Cog):
else:
raise BadArgument()
else:
time = None
if action.startswith("mute"):
time = time_pars(action.replace("mute", "")).total_seconds()
action = "mute"
elif action.startswith("ban"):
if action[3:]:
time = time_pars(action.replace("ban", "")).total_seconds()
action = "ban"
if time:
time = time_pars(time).total_seconds()
if a:
a.action = action
a.duration = time
@ -167,7 +175,7 @@ class Warn(commands.Cog):
s.commit()
s.close()
await ctx.message.add_reaction("\U0001f44d")
await ctx.send(content="\U0001f44d")
@commands.Cog.listener()
async def on_guild_remove(self, guild: Guild):

1
requirements.txt
View file

@ -3,6 +3,7 @@ async-timeout==3.0.1
attrs==20.2.0
chardet==3.0.4
discord==1.0.1
discord-py-slash-command==1.0.8.5
discord.py==1.5.1
feedparser==6.0.2
idna==2.10