Added first iteration of recruitmentCog
This commit is contained in:
parent
259a85a57e
commit
64af7ea6f8
5
recruitmentCog/__init__.py
Normal file
5
recruitmentCog/__init__.py
Normal file
@ -0,0 +1,5 @@
|
||||
from redbot.core.bot import Red
|
||||
from .recruitment import Recruitment
|
||||
|
||||
async def setup(bot: Red) -> None:
|
||||
await bot.add_cog(Recruitment(bot))
|
||||
315
recruitmentCog/recruitment.py
Normal file
315
recruitmentCog/recruitment.py
Normal file
@ -0,0 +1,315 @@
|
||||
import asyncio
|
||||
from typing import Union, List, Literal
|
||||
from datetime import timedelta
|
||||
from copy import copy
|
||||
import contextlib
|
||||
import discord
|
||||
|
||||
from redbot.core import Config, checks, commands
|
||||
from redbot.core.utils import AsyncIter
|
||||
from redbot.core.utils.chat_formatting import pagify, box
|
||||
from redbot.core.utils.antispam import AntiSpam
|
||||
from redbot.core.bot import Red
|
||||
from redbot.core.utils.predicates import MessagePredicate
|
||||
from redbot.core.utils.tunnel import Tunnel
|
||||
|
||||
|
||||
|
||||
class Recruitment(commands.Cog):
|
||||
"""Create user applications that server staff can respond to.
|
||||
|
||||
Users can open an application using `[p]apply`. These are then sent
|
||||
to a channel in the server for staff, and the application creator
|
||||
gets a DM. Both can be used to communicate.
|
||||
"""
|
||||
|
||||
default_guild_settings = {"output_channel": None, "active": False, "next_application": 1}
|
||||
|
||||
default_application = {"apply": {}}
|
||||
|
||||
# This can be made configurable later if it
|
||||
# becomes an issue.
|
||||
# Intervals should be a list of tuples in the form
|
||||
# (period: timedelta, max_frequency: int)
|
||||
# see redbot/core/utils/antispam.py for more details
|
||||
|
||||
intervals = [
|
||||
(timedelta(seconds=5), 1),
|
||||
(timedelta(minutes=5), 3),
|
||||
(timedelta(hours=1), 10),
|
||||
(timedelta(days=1), 24),
|
||||
]
|
||||
|
||||
def __init__(self, bot: Red):
|
||||
super().__init__()
|
||||
self.bot = bot
|
||||
self.antispam = {}
|
||||
self.user_cache = []
|
||||
self.tunnel_store = {}
|
||||
# (guild, ticket#):
|
||||
# {'tun': Tunnel, 'msgs': List[int]}
|
||||
|
||||
@property
|
||||
def tunnels(self):
|
||||
return [x["tun"] for x in self.tunnel_store.values()]
|
||||
|
||||
@checks.admin_or_permissions(manage_guild=True)
|
||||
@commands.guild_only()
|
||||
@commands.group(name="applicationset")
|
||||
async def applicationset(self, ctx: commands.Context):
|
||||
"""+++Manage Applications+++"""
|
||||
pass
|
||||
|
||||
@checks.admin_or_permissions(manage_guild=True)
|
||||
@applicationset.command(name="output")
|
||||
async def reportset_output(
|
||||
self, ctx: commands.Context, channel: Union[discord.TextChannel, discord.VoiceChannel]
|
||||
):
|
||||
"""Set the channel where applications will be sent."""
|
||||
await self.config.guild(ctx.guild).output_channel.set(channel.id)
|
||||
await ctx.send(_("The application channel has been set."))
|
||||
|
||||
@checks.admin_or_permissions(manage_guild=True)
|
||||
@applicationset.command(name="toggle", aliases=["toggleactive"])
|
||||
async def applicationset_toggle(self, ctx: commands.Context):
|
||||
"""Enable or disable applications for this server."""
|
||||
active = await self.config.guild(ctx.guild).active()
|
||||
active = not active
|
||||
await self.config.guild(ctx.guild).active.set(active)
|
||||
if active:
|
||||
await ctx.send(_("Applications are now enabled"))
|
||||
else:
|
||||
await ctx.send(_("Applications are now disabled"))
|
||||
|
||||
async def internal_filter(self, m: discord.Member, mod=False, perms=None):
|
||||
if perms and m.guild_permissions >= perms:
|
||||
return True
|
||||
if mod and await self.bot.is_mod(m):
|
||||
return True
|
||||
# The following line is for consistency with how perms are handled
|
||||
# in Red, though I'm not sure it makes sense to use here.
|
||||
if await self.bot.is_owner(m):
|
||||
return True
|
||||
|
||||
async def discover_guild(
|
||||
self,
|
||||
author: discord.User,
|
||||
*,
|
||||
mod: bool = False,
|
||||
permissions: Union[discord.Permissions, dict] = None,
|
||||
prompt: str = "",
|
||||
):
|
||||
"""
|
||||
discovers which of shared guilds between the bot
|
||||
and provided user based on conditions (mod or permissions is an or)
|
||||
|
||||
prompt is for providing a user prompt for selection
|
||||
"""
|
||||
shared_guilds = []
|
||||
if permissions is None:
|
||||
perms = discord.Permissions()
|
||||
elif isinstance(permissions, discord.Permissions):
|
||||
perms = permissions
|
||||
else:
|
||||
perms = discord.Permissions(**permissions)
|
||||
|
||||
async for guild in AsyncIter(self.bot.guilds, steps=100):
|
||||
x = guild.get_member(author.id)
|
||||
if x is not None:
|
||||
if await self.internal_filter(x, mod, perms):
|
||||
shared_guilds.append(guild)
|
||||
if len(shared_guilds) == 0:
|
||||
raise ValueError("No Qualifying Shared Guilds")
|
||||
if len(shared_guilds) == 1:
|
||||
return shared_guilds[0]
|
||||
output = ""
|
||||
guilds = sorted(shared_guilds, key=lambda g: g.name)
|
||||
for i, guild in enumerate(guilds, 1):
|
||||
output += "{}: {}\n".format(i, guild.name)
|
||||
output += "\n{}".format(prompt)
|
||||
|
||||
for page in pagify(output, delims=["\n"]):
|
||||
await author.send(box(page))
|
||||
|
||||
try:
|
||||
message = await self.bot.wait_for(
|
||||
"message",
|
||||
check=MessagePredicate.same_context(channel=author.dm_channel, user=author),
|
||||
timeout=45,
|
||||
)
|
||||
except asyncio.TimeoutError:
|
||||
await author.send(_("You took too long to select. Try again later."))
|
||||
return None
|
||||
|
||||
try:
|
||||
message = int(message.content.strip())
|
||||
guild = guilds[message - 1]
|
||||
except (ValueError, IndexError):
|
||||
await author.send(_("That wasn't a valid choice."))
|
||||
return None
|
||||
else:
|
||||
return guild
|
||||
|
||||
async def send_application(self, ctx: commands.Context, msg: discord.Message, guild: discord.Guild):
|
||||
author = guild.get_member(msg.author.id)
|
||||
application = msg.clean_content
|
||||
|
||||
channel_id = await self.config.guild(guild).output_channel()
|
||||
channel = guild.get_channel(channel_id)
|
||||
if channel is None:
|
||||
return None
|
||||
|
||||
files: List[discord.File] = await Tunnel.files_from_attach(msg)
|
||||
|
||||
application_number = await self.config.guild(guild).next_application()
|
||||
await self.config.guild(guild).next_application.set(application_number + 1)
|
||||
|
||||
if await self.bot.embed_requested(channel):
|
||||
em = discord.Embed(description=application, colour=await ctx.embed_colour())
|
||||
em.set_author(
|
||||
name=_("Application from {author}{maybe_nick}").format(
|
||||
author=author, maybe_nick=(f" ({author.nick})" if author.nick else "")
|
||||
),
|
||||
icon_url=author.display_avatar,
|
||||
)
|
||||
em.set_footer(text=_("Application #{}").format(application_number))
|
||||
send_content = None
|
||||
else:
|
||||
em = None
|
||||
send_content = _("Application from {author.mention} (Application #{number})").format(
|
||||
author=author, number=application_number
|
||||
)
|
||||
send_content += "\n" + application
|
||||
|
||||
try:
|
||||
await Tunnel.message_forwarder(
|
||||
destination=channel, content=send_content, embed=em, files=files
|
||||
)
|
||||
except (discord.Forbidden, discord.HTTPException):
|
||||
return None
|
||||
|
||||
return application_number
|
||||
|
||||
@commands.group(name="Application", usage="[text]", invoke_without_command=True)
|
||||
async def application(self, ctx: commands.Context, *, _application: str = ""):
|
||||
"""Send an application.
|
||||
|
||||
Use without arguments for an interactive experience, or do
|
||||
`[p]apply [text]` to use it non-interactively.
|
||||
"""
|
||||
author = ctx.author
|
||||
guild = ctx.guild
|
||||
if guild is None:
|
||||
guild = await self.discover_guild(
|
||||
author, prompt=_("Select a server to make an application in by number.")
|
||||
)
|
||||
if guild is None:
|
||||
return
|
||||
g_active = await self.config.guild(guild).active()
|
||||
if not g_active:
|
||||
return await author.send(_("Applications are currently closed"))
|
||||
if guild.id not in self.antispam:
|
||||
self.antispam[guild.id] = {}
|
||||
if author.id not in self.antispam[guild.id]:
|
||||
self.antispam[guild.id][author.id] = AntiSpam(self.intervals)
|
||||
if self.antispam[guild.id][author.id].spammy:
|
||||
return await author.send(
|
||||
_(
|
||||
"You've sent too many applications recently. "
|
||||
"Are you sure you are in the right place? "
|
||||
)
|
||||
)
|
||||
if author.id in self.user_cache:
|
||||
return await author.send(
|
||||
_(
|
||||
"Please finish making your prior application before trying to make an "
|
||||
"additional one!"
|
||||
)
|
||||
)
|
||||
self.user_cache.append(author.id)
|
||||
|
||||
if _application:
|
||||
_m = copy(ctx.message)
|
||||
_m.content = _application
|
||||
_m.content = _m.clean_content
|
||||
val = await self.send_application(ctx, _m, guild)
|
||||
else:
|
||||
try:
|
||||
await author.send(
|
||||
_(
|
||||
"Please respond to this message with your application."
|
||||
"\nYour application should be a single message"
|
||||
)
|
||||
)
|
||||
except discord.Forbidden:
|
||||
return await ctx.send(_("This requires DMs enabled."))
|
||||
|
||||
try:
|
||||
message = await self.bot.wait_for(
|
||||
"message",
|
||||
check=MessagePredicate.same_context(ctx, channel=author.dm_channel),
|
||||
timeout=180,
|
||||
)
|
||||
except asyncio.TimeoutError:
|
||||
return await author.send(_("You took too long. Try again later."))
|
||||
else:
|
||||
val = await self.send_application(ctx, message, guild)
|
||||
|
||||
with contextlib.suppress(discord.Forbidden, discord.HTTPException):
|
||||
if val is None:
|
||||
if await self.config.guild(ctx.guild).output_channel() is None:
|
||||
await author.send(
|
||||
_(
|
||||
"Hmm, most embarrassing, it rather seems Hatt has neglected to tell me where the applications room is. Please contact him for me."
|
||||
)
|
||||
)
|
||||
else:
|
||||
await author.send(
|
||||
_("Drat! There was an error sending your application, please contact Hatt.")
|
||||
)
|
||||
else:
|
||||
await author.send(_("Your application was submitted. (Application #{})").format(val))
|
||||
self.antispam[guild.id][author.id].stamp()
|
||||
|
||||
@application.after_invoke
|
||||
async def application_cleanup(self, ctx: commands.Context):
|
||||
"""
|
||||
The logic is cleaner this way
|
||||
"""
|
||||
if ctx.author.id in self.user_cache:
|
||||
self.user_cache.remove(ctx.author.id)
|
||||
if ctx.guild and ctx.invoked_subcommand is None:
|
||||
if ctx.bot_permissions.manage_messages:
|
||||
try:
|
||||
await ctx.message.delete()
|
||||
except discord.NotFound:
|
||||
pass
|
||||
|
||||
@commands.Cog.listener()
|
||||
async def on_message(self, message: discord.Message):
|
||||
to_remove = []
|
||||
|
||||
for k, v in self.tunnel_store.items():
|
||||
guild, application_number = k
|
||||
if await self.bot.cog_disabled_in_guild(self, guild):
|
||||
to_remove.append(k)
|
||||
continue
|
||||
|
||||
topic = _("Re: application# {application_number} in {guild.name}").format(
|
||||
application_number=application_number, guild=guild
|
||||
)
|
||||
# Tunnels won't forward unintended messages, this is safe
|
||||
msgs = await v["tun"].communicate(message=message, topic=topic)
|
||||
if msgs:
|
||||
self.tunnel_store[k]["msgs"] = msgs
|
||||
|
||||
for key in to_remove:
|
||||
if tun := self.tunnel_store.pop(key, None):
|
||||
guild, application = key
|
||||
await tun["tun"].close_because_disabled(
|
||||
_(
|
||||
"Correspondence about application# {application_number} in "
|
||||
"{guild.name} has been ended due "
|
||||
"to applications being closed."
|
||||
).format(application_number=application, guild=guild)
|
||||
)
|
||||
Loading…
x
Reference in New Issue
Block a user