Strip this down to only what we need.

main
Nichole Mattera 3 years ago
parent b805ca15c2
commit 140f6d8299
  1. 100
      README.md
  2. 51
      Robocop.py
  3. 52
      cogs/admin.py
  4. 47
      cogs/basic.py
  5. 135
      cogs/common.py
  6. 196
      cogs/err.py
  7. 43
      cogs/invites.py
  8. 94
      cogs/links.py
  9. 466
      cogs/lists.py
  10. 263
      cogs/lists_verification.py
  11. 90
      cogs/lockdown.py
  12. 353
      cogs/logs.py
  13. 227
      cogs/meme.py
  14. 551
      cogs/mod.py
  15. 143
      cogs/mod_mail.py
  16. 32
      cogs/mod_note.py
  17. 110
      cogs/mod_reacts.py
  18. 160
      cogs/mod_stats.py
  19. 137
      cogs/mod_timed.py
  20. 221
      cogs/mod_userlog.py
  21. 46
      cogs/mod_watch.py
  22. 157
      cogs/pin.py
  23. 68
      cogs/remind.py
  24. 153
      cogs/robocronp.py
  25. 52
      cogs/uwu.py
  26. 123
      config_template.py
  27. 17
      helpers/checks.py
  28. 1115
      helpers/errcodes.py
  29. 42
      helpers/restrictions.py
  30. 37
      helpers/robocronp.py
  31. 72
      helpers/userlogs.py

@ -1,4 +1,4 @@
# Komet-CL (Crash Landing)
# Dominic Botoretto
Next-gen rewrite of Kurisu/Robocop/Komet bot used by ~~AtlasNX~~, ReSwitched and Nintendo Homebrew with discord.py rewrite, designed to be relatively clean, consistent and un-bloated.
@ -29,104 +29,6 @@ If you're moving from Kurisu/Robocop, and want to preserve your data, you'll wan
---
## TODO
All Robocop features are now supported.
<details>
<summary>List of added Kurisu/Robocop features</summary>
<p>
- [x] .py configs
- [x] membercount command
- [x] Meme commands and pegaswitch (honestly the easiest part)
- [x] source command
- [x] robocop command
- [x] Verification: Actual verification system
- [x] Verification: Reset command
- [x] Logging: joins
- [x] Logging: leaves
- [x] Logging: role changes
- [x] Logging: bans
- [x] Logging: kicks
- [x] Moderation: speak
- [x] Moderation: ban
- [x] Moderation: silentban
- [x] Moderation: kick
- [x] Moderation: userinfo
- [x] Moderation: approve-revoke (community)
- [x] Moderation: addhacker-removehacker (hacker)
- [x] Moderation: probate-unprobate (participant)
- [x] Moderation: lock-softlock-unlock (channel lockdown)
- [x] Moderation: mute-unmute
- [x] Moderation: playing
- [x] Moderation: botnickname
- [x] Moderation: nickname
- [x] Moderation: clear/purge
- [x] Moderation: restrictions (people who leave with muted role will get muted role on join)
- [x] Warns: warn
- [x] Warns: listwarns-listwarnsid
- [x] Warns: clearwarns-clearwarnsid
- [x] Warns: delwarnid-delwarn
- [x] .serr and .err (thanks tomger!)
</p>
</details>
---
The main goal of this project, to get Robocop functionality done, is complete.
Secondary goal is adding new features:
- [ ] New feature: Submiterr (relies on modmail)
- [ ] Feature creep: Shortlink completion (gl/ao/etc)
- [ ] New moderation feature: timelock (channel lockdown with time, relies on robocronp)
<details>
<summary>Completed features</summary>
<p>
- [x] Better security, better checks and better guild whitelisting
- [x] Feature creep: Reminds
- [x] A system for running jobs in background with an interval (will be called robocronp)
- [x] Commands to list said jobs and remove them
- [x] New moderation feature: timemute (mute with time, relies on robocronp)
- [x] New moderation feature: timeban (ban with expiry, relies on robocronp)
- [x] Improvements to lockdown to ensure that staff can talk
- [x] New moderation feature: Display of mutes, bans and kicks on listwarns (.userlog now)
- [x] New moderation feature: User notes
- [x] New moderation feature: Reaction removing features (thanks misson20000!)
- [x] New moderation feature: User nickname change
- [x] New moderation feature: watch-unwatch
- [x] New moderation feature: tracking suspicious keywords
- [x] New moderation feature: tracking invites posted
- [x] New self-moderation feature: .mywarns
- [x] New feature: Highlights (problematic words automatically get posted to modmail channel, relies on modmail)
- [x] Purge: On purge, send logs in form of txt file to server logs
- [x] New feature: Modmail
</p>
</details>
<details>
<summary>TODO for robocronp</summary>
<p>
- [ ] Reduce code repetition on mod_timed.py
- [x] Allow non-hour values on timed bans
the following require me to rethink some of the lockdown code, which I don't feel like
- [ ] lockdown in helper
- [ ] timelock command
- [ ] working cronjob for unlock
</p>
</details>
---
## Credits
Robocop-NG is currently developed and maintained by @aveao and @tumGER. The official bot is hosted by @yuukieve.

@ -12,23 +12,17 @@ from discord.ext import commands
script_name = os.path.basename(__file__).split('.')[0]
log_file_name = f"{script_name}.log"
# Limit of discord (non-nitro) is 8MB (not MiB)
max_file_size = 1000 * 1000 * 8
backup_count = 3
file_handler = logging.handlers.RotatingFileHandler(
filename=log_file_name, maxBytes=max_file_size, backupCount=backup_count)
stdout_handler = logging.StreamHandler(sys.stdout)
log_format = logging.Formatter(
'[%(asctime)s] {%(filename)s:%(lineno)d} %(levelname)s - %(message)s')
file_handler.setFormatter(log_format)
stdout_handler.setFormatter(log_format)
log = logging.getLogger('discord')
log.setLevel(logging.INFO)
log.addHandler(file_handler)
log.addHandler(stdout_handler)
@ -37,32 +31,11 @@ def get_prefix(bot, message):
return commands.when_mentioned_or(*prefixes)(bot, message)
wanted_jsons = ["data/restrictions.json",
"data/robocronptab.json",
"data/userlog.json",
"data/invites.json"]
initial_extensions = ['cogs.common',
'cogs.admin',
'cogs.mod',
'cogs.mod_mail',
'cogs.mod_note',
'cogs.mod_reacts',
'cogs.mod_stats',
'cogs.mod_userlog',
'cogs.mod_timed',
'cogs.mod_watch',
'cogs.basic',
'cogs.logs',
'cogs.lockdown',
'cogs.links',
'cogs.lists',
'cogs.lists_verification',
'cogs.remind',
'cogs.robocronp',
'cogs.meme',
'cogs.uwu']
'cogs.meme']
bot = commands.Bot(command_prefix=get_prefix,
description=config.bot_description, pm_help=True)
@ -71,7 +44,6 @@ bot.log = log
bot.loop = asyncio.get_event_loop()
bot.config = config
bot.script_name = script_name
bot.wanted_jsons = wanted_jsons
if __name__ == '__main__':
for extension in initial_extensions:
@ -95,11 +67,7 @@ async def on_ready():
# Send "Robocop has started! x has y members!"
guild = bot.botlog_channel.guild
msg = f"{bot.user.name} has started! "\
f"{guild.name} has {guild.member_count} members!"
data_files = [discord.File(fpath) for fpath in wanted_jsons]
await bot.botlog_channel.send(msg, files=data_files)
msg = f"{bot.user.name} has started!"
activity = discord.Activity(name=game_name,
type=discord.ActivityType.listening)
@ -190,22 +158,7 @@ async def on_message(message):
if (message.guild) and (message.guild.id not in config.guild_whitelist):
return
# Ignore messages in newcomers channel, unless it's potentially
# an allowed command
welcome_allowed = ["reset", "kick", "ban", "warn"]
if message.channel.id == config.welcome_channel and\
not any(cmd in message.content for cmd in welcome_allowed):
return
ctx = await bot.get_context(message)
await bot.invoke(ctx)
if not os.path.exists("data"):
os.makedirs("data")
for wanted_json in wanted_jsons:
if not os.path.exists(wanted_json):
with open(wanted_json, "w") as f:
f.write("{}")
bot.run(config.token, bot=True, reconnect=True)

@ -2,11 +2,8 @@ import discord
from discord.ext import commands
from discord.ext.commands import Cog
import traceback
import inspect
import re
import config
from helpers.checks import check_if_bot_manager
from helpers.userlogs import set_userlog
class Admin(Cog):
@ -14,6 +11,11 @@ class Admin(Cog):
self.bot = bot
self.last_eval_result = None
self.previous_eval_code = None
async def cog_load_actions(self, cog_name):
if cog_name == "verification":
verif_channel = self.bot.get_channel(config.welcome_channel)
await self.bot.do_resetalgo(verif_channel, "cog load")
@commands.guild_only()
@commands.check(check_if_bot_manager)
@ -23,33 +25,6 @@ class Admin(Cog):
await ctx.send(":wave: Goodbye!")
await self.bot.logout()
@commands.guild_only()
@commands.check(check_if_bot_manager)
@commands.command()
async def fetchlog(self, ctx):
"""Returns log"""
await ctx.send("Here's the current log file:",
file=discord.File(f"{self.bot.script_name}.log"))
@commands.guild_only()
@commands.check(check_if_bot_manager)
@commands.command()
async def fetchdata(self, ctx):
"""Returns data files"""
data_files = [discord.File(fpath) for fpath in self.bot.wanted_jsons]
await ctx.send("Here you go:", files=data_files)
@commands.guild_only()
@commands.check(check_if_bot_manager)
@commands.command(name='eval')
async def _eval(self, ctx):
await ctx.send("Fuck off. This doesn't belong in production code!")
async def cog_load_actions(self, cog_name):
if cog_name == "verification":
verif_channel = self.bot.get_channel(config.welcome_channel)
await self.bot.do_resetalgo(verif_channel, "cog load")
@commands.guild_only()
@commands.check(check_if_bot_manager)
@commands.command()
@ -117,22 +92,5 @@ class Admin(Cog):
self.bot.log.info(f'Reloaded ext {ext}')
await ctx.send(f':white_check_mark: `{ext}` successfully reloaded.')
@commands.guild_only()
@commands.check(check_if_bot_manager)
@commands.command()
async def restoreuserlog(self, ctx, message_id: int):
"""Restores the user log from a saved backup."""
message = await self.bot.botlog_channel.fetch_message(message_id)
if message.author.id != self.bot.user.id or len(message.attachments) == 0:
await ctx.send("Incorrect Message ID.")
for attachment in message.attachments:
if attachment.filename == "userlog.json":
logs = (await attachment.read()).decode("utf-8")
set_userlog(logs)
return
await ctx.send("Incorrect Message ID.")
def setup(bot):
bot.add_cog(Admin(bot))

@ -3,64 +3,19 @@ import config
import discord
from discord.ext import commands
from discord.ext.commands import Cog
from helpers.checks import check_if_verified
class Basic(Cog):
def __init__(self, bot):
self.bot = bot
@commands.guild_only()
@commands.check(check_if_verified)
@commands.command()
async def hello(self, ctx):
"""Says hello. Duh."""
await ctx.send(f"Hello {ctx.author.mention}!")
@commands.check(check_if_verified)
@commands.command(aliases=['aboutkosmos'])
async def about(self, ctx):
"""Shows what kosmos is and what it includes"""
await ctx.send("Kosmos is a CFW bundle that comes with Atmosphere, Hekate, and some homebrew. You can see all the homebrew that is included here: https://github.com/AtlasNX/Kosmos#featuring")
@commands.check(check_if_verified)
@commands.command(aliases=["fat32"])
async def exfat(self, ctx):
"""Displays a helpful message on why not to use exFAT"""
embed = discord.Embed(title="GUIFormat",
url="http://www.ridgecrop.demon.co.uk/guiformat.exe",
description="A useful tool for formatting SD cards over 32GB as FAT32 on Windows.")
message_text=("The exFAT drivers built into the Switch has been known "
"to corrupt SD cards and homebrew only makes this worse. "
"Backup everything on your SD card as soon as possible "
"and format it to FAT32. On Windows, if your SD card is "
"over 32GB then it will not let you select FAT32 from "
"the built-in format tool, however you can use a tool "
"like GUIFormat to format it.")
await ctx.send(content=message_text,
embed=embed)
@commands.guild_only()
@commands.check(check_if_verified)
@commands.command()
async def membercount(self, ctx):
"""Prints the member count of the server."""
await ctx.send(f"{ctx.guild.name} has "
f"{ctx.guild.member_count} members!")
@commands.check(check_if_verified)
@commands.command(aliases=["robocopng", "robocop-ng", "komet", "komet-cl"])
async def robocop(self, ctx):
"""Shows a quick embed with bot info."""
embed = discord.Embed(title="Komet",
url=config.source_url,
description=config.embed_desc)
embed.set_thumbnail(url=self.bot.user.avatar_url)
await ctx.send(embed=embed)
@commands.check(check_if_verified)
@commands.command(aliases=['p', "ddos"])
@commands.command(aliases=['p'])
async def ping(self, ctx):
"""Shows ping values to discord.

@ -12,144 +12,13 @@ class Common(Cog):
self.bot = bot
self.bot.async_call_shell = self.async_call_shell
self.bot.slice_message = self.slice_message
self.max_split_length = 3
self.bot.hex_to_int = self.hex_to_int
self.bot.download_file = self.download_file
self.bot.aiojson = self.aiojson
self.bot.aioget = self.aioget
self.bot.aiogetbytes = self.aiogetbytes
self.bot.get_relative_timestamp = self.get_relative_timestamp
self.bot.escape_message = self.escape_message
self.bot.parse_time = self.parse_time
self.bot.haste = self.haste
def parse_time(self, delta_str):
cal = parsedatetime.Calendar()
time_struct, parse_status = cal.parse(delta_str)
res_timestamp = math.floor(time.mktime(time_struct))
return res_timestamp
def get_relative_timestamp(self, time_from=None, time_to=None,
humanized=False, include_from=False,
include_to=False):
# Setting default value to utcnow() makes it show time from cog load
# which is not what we want
if not time_from:
time_from = datetime.datetime.utcnow()
if not time_to:
time_to = datetime.datetime.utcnow()
if humanized:
humanized_string = humanize.naturaltime(time_from - time_to)
if include_from and include_to:
str_with_from_and_to = f"{humanized_string} "\
f"({str(time_from).split('.')[0]} "\
f"- {str(time_to).split('.')[0]})"
return str_with_from_and_to
elif include_from:
str_with_from = f"{humanized_string} "\
f"({str(time_from).split('.')[0]})"
return str_with_from
elif include_to:
str_with_to = f"{humanized_string} "\
f"({str(time_to).split('.')[0]})"
return str_with_to
return humanized_string
else:
epoch = datetime.datetime.utcfromtimestamp(0)
epoch_from = (time_from - epoch).total_seconds()
epoch_to = (time_to - epoch).total_seconds()
second_diff = epoch_to - epoch_from
result_string = str(datetime.timedelta(
seconds=second_diff)).split('.')[0]
return result_string
async def aioget(self, url):
try:
data = await self.bot.aiosession.get(url)
if data.status == 200:
text_data = await data.text()
self.bot.log.info(f"Data from {url}: {text_data}")
return text_data
else:
self.bot.log.error(f"HTTP Error {data.status} "
"while getting {url}")
except:
self.bot.log.error(f"Error while getting {url} "
f"on aiogetbytes: {traceback.format_exc()}")
async def aiogetbytes(self, url):
try:
data = await self.bot.aiosession.get(url)
if data.status == 200:
byte_data = await data.read()
self.bot.log.debug(f"Data from {url}: {byte_data}")
return byte_data
else:
self.bot.log.error(f"HTTP Error {data.status} "
"while getting {url}")
except:
self.bot.log.error(f"Error while getting {url} "
f"on aiogetbytes: {traceback.format_exc()}")
async def aiojson(self, url):
try:
data = await self.bot.aiosession.get(url)
if data.status == 200:
text_data = await data.text()
self.bot.log.info(f"Data from {url}: {text_data}")
content_type = data.headers['Content-Type']
return await data.json(content_type=content_type)
else:
self.bot.log.error(f"HTTP Error {data.status} "
"while getting {url}")
except:
self.bot.log.error(f"Error while getting {url} "
f"on aiogetbytes: {traceback.format_exc()}")
def hex_to_int(self, color_hex: str):
"""Turns a given hex color into an integer"""
return int("0x" + color_hex.strip('#'), 16)
def escape_message(self, text: str):
"""Escapes unfun stuff from messages"""
"""Escapes unfun stuff from messages."""
return str(text).replace("@", "@ ").replace("<#", "# ")
# This function is based on https://stackoverflow.com/a/35435419/3286892
# by link2110 (https://stackoverflow.com/users/5890923/link2110)
# modified by Ave (https://github.com/aveao), licensed CC-BY-SA 3.0
async def download_file(self, url, local_filename):
file_resp = await self.bot.aiosession.get(url)
file = await file_resp.read()
with open(local_filename, "wb") as f:
f.write(file)
# 2000 is maximum limit of discord
async def slice_message(self, text, size=2000, prefix="", suffix=""):
"""Slices a message into multiple messages"""
if len(text) > size * self.max_split_length:
haste_url = await self.haste(text)
return [f"Message is too long ({len(text)} > "
f"{size * self.max_split_length} "
f"({size} * {self.max_split_length}))"
f", go to haste: <{haste_url}>"]
reply_list = []
size_wo_fix = size - len(prefix) - len(suffix)
while len(text) > size_wo_fix:
reply_list.append(f"{prefix}{text[:size_wo_fix]}{suffix}")
text = text[size_wo_fix:]
reply_list.append(f"{prefix}{text}{suffix}")
return reply_list
async def haste(self, text, instance='https://mystb.in/'):
response = await self.bot.aiosession.post(f"{instance}documents",
data=text)
if response.status == 200:
result_json = await response.json()
return f"{instance}{result_json['key']}"
else:
return f"Error {response.status}: {response.text}"
async def async_call_shell(self, shell_command: str,
inc_stdout=True, inc_stderr=True):
pipe = asyncio.subprocess.PIPE

@ -1,196 +0,0 @@
import re
import discord
from discord.ext import commands
from discord.ext.commands import Cog
from helpers.checks import check_if_verified
from helpers.errcodes import *
class Err(Cog):
"""Everything related to Nintendo 3DS, Wii U and Switch error codes"""
def __init__(self, bot):
self.bot = bot
self.dds_re = re.compile(r'0\d{2}\-\d{4}')
self.wiiu_re = re.compile(r'1\d{2}\-\d{4}')
self.switch_re = re.compile(r'2\d{3}\-\d{4}')
self.no_err_desc = "It seems like your error code is unknown. "\
"You should report relevant details to "\
"<@141532589725974528> (tomGER#7462) "\
"so it can be added to the bot."
self.rickroll = "https://www.youtube.com/watch?v=4uj896lr3-E"
@commands.check(check_if_verified)
@commands.command(aliases=["3dserr", "3err", "dserr"])
async def dderr(self, ctx, err: str):
"""Searches for 3DS error codes!
Usage: .ddserr/.3err/.dserr/.3dserr <Error Code>"""
if self.dds_re.match(err): # 3DS - dds -> Drei DS -> Three DS
if err in dds_errcodes:
err_description = dds_errcodes[err]
else:
err_description = self.no_err_desc
# Make a nice Embed out of it
embed = discord.Embed(title=err,
url=self.rickroll,
description=err_description)
embed.set_footer(text="Console: 3DS")
# Send message, crazy
await ctx.send(embed=embed)
# These are not similar to the other errors apparently ... ?
elif err.startswith("0x"):
derr = err[2:]
derr = derr.strip()
rc = int(derr, 16)
desc = rc & 0x3FF
mod = (rc >> 10) & 0xFF
summ = (rc >> 21) & 0x3F
level = (rc >> 27) & 0x1F
embed = discord.Embed(title=f"0x{rc:X}")
embed.add_field(name="Module", value=dds_modules.get(mod, mod))
embed.add_field(name="Description",
value=dds_descriptions.get(desc, desc))
embed.add_field(name="Summary", value=dds_summaries.get(summ, summ))
embed.add_field(name="Level", value=dds_levels.get(level, level))
embed.set_footer(text="Console: 3DS")
await ctx.send(embed=embed)
return
else:
await ctx.send("Unknown Format - This is either "
"no error code or you made some mistake!")
@commands.check(check_if_verified)
@commands.command(aliases=["wiiuserr", "uerr", "wuerr", "mochaerr"])
async def wiiuerr(self, ctx, err: str):
"""Searches for Wii U error codes!
Usage: .wiiuserr/.uerr/.wuerr/.mochaerr <Error Code>"""
if self.wiiu_re.match(err): # Wii U
module = err[2:3] # Is that even true, idk just guessing
desc = err[5:8]
if err in wii_u_errors:
err_description = wii_u_errors[err]
else:
err_description = self.no_err_desc
# Make a nice Embed out of it
embed = discord.Embed(title=err,
url=self.rickroll,
description=err_description)
embed.set_footer(text="Console: Wii U")
embed.add_field(name="Module", value=module, inline=True)
embed.add_field(name="Description", value=desc, inline=True)
# Send message, crazy
await ctx.send(embed=embed)
else:
await ctx.send("Unknown Format - This is either "
"no error code or you made some mistake!")
@commands.check(check_if_verified)
@commands.command(aliases=["nxerr", "serr"])
async def err(self, ctx, err: str):
"""Searches for Switch error codes!
Usage: .serr/.nxerr/.err <Error Code>"""
if self.switch_re.match(err) or err.startswith("0x"): # Switch
if err.startswith("0x"):
err = err[2:]
errcode = int(err, 16)
module = errcode & 0x1FF
desc = (errcode >> 9) & 0x3FFF
else:
module = int(err[0:4]) - 2000
desc = int(err[5:9])
errcode = (desc << 9) + module
str_errcode = f'{(module + 2000):04}-{desc:04}'
# Searching for Modules in list
if module in switch_modules:
err_module = switch_modules[module]
else:
err_module = "Unknown"
# Set initial value unconditionally
err_description = self.no_err_desc
# Searching for error codes related to the Switch
# (doesn't include special cases)
if errcode in switch_known_errcodes:
err_description = switch_known_errcodes[errcode]
elif errcode in switch_support_page:
err_description = switch_support_page[errcode]
elif module in switch_known_errcode_ranges:
for errcode_range in switch_known_errcode_ranges[module]:
if desc >= errcode_range[0] and desc <= errcode_range[1]:
err_description = errcode_range[2]
# Make a nice Embed out of it
embed = discord.Embed(title=f"{str_errcode} / {hex(errcode)}",
url=self.rickroll,
description=err_description)
embed.add_field(name="Module",
value=f"{err_module} ({module})",
inline=True)
embed.add_field(name="Description", value=desc, inline=True)
if "ban" in err_description:
embed.set_footer("F to you | Console: Switch")
else:
embed.set_footer(text="Console: Switch")
await ctx.send(embed=embed)
# Special case handling because Nintendo feels like
# its required to break their format lol
elif err in switch_game_err:
game, desc = switch_game_err[err].split(":")
embed = discord.Embed(title=err,
url=self.rickroll,
description=desc)
embed.set_footer(text="Console: Switch")
embed.add_field(name="Game", value=game, inline=True)
await ctx.send(embed=embed)
else:
await ctx.send("Unknown Format - This is either "
"no error code or you made some mistake!")
@commands.check(check_if_verified)
@commands.command(aliases=["e2h"])
async def err2hex(self, ctx, err: str):
"""Converts Nintendo Switch errors to hex
Usage: .err2hex <Error Code>"""
if self.switch_re.match(err):
module = int(err[0:4]) - 2000
desc = int(err[5:9])
errcode = (desc << 9) + module
await ctx.send(hex(errcode))
else:
await ctx.send("This doesn't follow the typical"
" Nintendo Switch 2XXX-XXXX format!")
@commands.check(check_if_verified)
@commands.command(aliases=["h2e"])
async def hex2err(self, ctx, err: str):
"""Converts Nintendo Switch errors to hex
Usage: .hex2err <Hex>"""
if err.startswith("0x"):
err = err[2:]
err = int(err, 16)
module = err & 0x1FF
desc = (err >> 9) & 0x3FFF
errcode = f'{(module + 2000):04}-{desc:04}'
await ctx.send(errcode)
else:
await ctx.send("This doesn't look like typical hex!")
def setup(bot):
bot.add_cog(Err(bot))

@ -1,43 +0,0 @@
from discord.ext import commands
from discord.ext.commands import Cog
from helpers.checks import check_if_collaborator
import config
import json
class Invites(Cog):
def __init__(self, bot):
self.bot = bot
@commands.command()
@commands.guild_only()
@commands.check(check_if_collaborator)
async def invite(self, ctx):
welcome_channel = self.bot.get_channel(config.welcome_channel)
author = ctx.message.author
reason = f"Created by {str(author)} ({author.id})"
invite = await welcome_channel.create_invite(max_age = 0,
max_uses = 1, temporary = True, unique = True, reason = reason)
with open("data/invites.json", "r") as f:
invites = json.load(f)
invites[invite.id] = {
"uses": 0,
"url": invite.url,
"max_uses": 1,
"code": invite.code
}
with open("data/invites.json", "w") as f:
f.write(json.dumps(invites))
await ctx.message.add_reaction("๐Ÿ†—")
try:
await ctx.author.send(f"Created single-use invite {invite.url}")
except discord.errors.Forbidden:
await ctx.send(f"{ctx.author.mention} I could not send you the \
invite. Send me a DM so I can reply to you.")
def setup(bot):
bot.add_cog(Invites(bot))

@ -2,104 +2,24 @@ import discord
import config
from discord.ext import commands
from discord.ext.commands import Cog
from helpers.checks import check_if_verified
class Links(Cog):
"""
Commands for easily linking to projects.
Commands for easily linking to stuff.
"""
def __init__(self, bot):
self.bot = bot
@commands.check(check_if_verified)
@commands.command(hidden=True)
async def pegaswitch(self, ctx):
"""Link to the Pegaswitch repo"""
await ctx.send("https://github.com/reswitched/pegaswitch")
@commands.command(aliases=["video"])
async def zoom(self, ctx):
"""Link to the Family Zoom Room."""
await ctx.send(config.zoom_url)
@commands.check(check_if_verified)
@commands.command(hidden=True, aliases=["atmos", "ams"])
async def atmosphere(self, ctx):
"""Link to the Atmosphere repo"""
await ctx.send("https://github.com/atmosphere-nx/atmosphere")
@commands.check(check_if_verified)
@commands.command(hidden=True, aliases=["bootloader"])
async def hekate(self, ctx):
"""Link to the Hekate repo"""
await ctx.send("https://github.com/CTCaer/hekate")
@commands.check(check_if_verified)
@commands.command(hidden=True, aliases=["xyproblem"])
async def xy(self, ctx):
"""Link to the "What is the XY problem?" post from SE"""
await ctx.send("<https://meta.stackexchange.com/q/66377/285481>\n\n"
"TL;DR: It's asking about your attempted solution "
"rather than your actual problem.\n"
"It's perfectly okay to want to learn about a "
"solution, but please be clear about your intentions "
"if you're not actually trying to solve a problem.")
@commands.check(check_if_verified)
@commands.command(hidden=True, aliases=["guides", "link"])
async def guide(self, ctx):
"""Link to the guide(s)"""
message_text=("**Generic starter guides:**\n"
"AtlasNX's Guide: "
"<https://switch.homebrew.guide>\n"
"\n"
"**Specific guides:**\n"
"Manually Updating/Downgrading (with HOS): "
"<https://switch.homebrew.guide/usingcfw/manualupgrade>\n"
"Manually Repairing/Downgrading (without HOS): "
"<https://switch.homebrew.guide/usingcfw/manualchoiupgrade>\n"
"How to get started developing Homebrew: "
"<https://switch.homebrew.guide/homebrew_dev/introduction>\n"
"\n")
try:
support_faq_channel = self.bot.get_channel(config.support_faq_channel)
if support_faq_channel is None:
message_text += "Check out #support-faq for additional help."
else:
message_text += f"Check out {support_faq_channel.mention} for additional help."
except AttributeError:
message_text += "Check out #support-faq for additional help."
await ctx.send(message_text)
@commands.check(check_if_verified)
@commands.command(hidden=True, aliases=["patron"])
async def patreon(self, ctx):
"""Link to the patreon"""
await ctx.send("https://patreon.teamatlasnx.com")
@commands.check(check_if_verified)
@commands.command(hidden=True, aliases=["coffee"])
async def kofi(self, ctx):
"""Link to Ko-fi"""
await ctx.send("https://kofi.teamatlasnx.com")
@commands.check(check_if_verified)
@commands.command(hidden=True, aliases=["sdfiles"])
async def kosmos(self, ctx):
"""Link to the latest Kosmos release"""
await ctx.send("https://github.com/AtlasNX/Kosmos/releases/latest")
@commands.check(check_if_verified)
@commands.command(hidden=True, aliases=["sd"])
async def sdsetup(self, ctx):
"""Link to SD Setup"""
await ctx.send("https://sdsetup.com")
@commands.check(check_if_verified)
@commands.command()
async def source(self, ctx):
"""Gives link to source code."""
await ctx.send(f"You can find my source at {config.source_url}. "
"Serious PRs and issues welcome!")
"""Link to the Dominic Botoretto source code."""
await ctx.send(config.source_url)
def setup(bot):
bot.add_cog(Links(bot))

@ -1,466 +0,0 @@
import config
import discord
from discord.ext import commands
from discord.ext.commands import Cog
from helpers.checks import check_if_staff, check_if_verified
import io
import os
import urllib.parse
class Lists(Cog):
"""
Manages channels that are dedicated to lists.
"""
def __init__(self, bot):
self.bot = bot
# Helpers
def check_if_target_is_staff(self, target):
return any(r.id in config.staff_role_ids for r in target.roles)
def is_edit(self, emoji):
return str(emoji)[0] == "โœ" or str(emoji)[0] == "๐Ÿ“"
def is_delete(self, emoji):
return str(emoji)[0] == "โŒ" or str(emoji)[0] == "โŽ"
def is_recycle(self, emoji):
return str(emoji)[0] == "โ™ป"
def is_insert_above(self, emoji):
return str(emoji)[0] == "โคด๏ธ" or str(emoji)[0] == "โฌ†"
def is_insert_below(self, emoji):
return str(emoji)[0] == "โคต๏ธ" or str(emoji)[0] == "โฌ‡"
def is_reaction_valid(self, reaction):
allowed_reactions = [
"โœ",
"๐Ÿ“",
"โŒ",
"โŽ",
"โ™ป",
"โคด๏ธ",
"โฌ†",
"โฌ‡",
"โคต๏ธ",
]
return str(reaction.emoji)[0] in allowed_reactions
async def find_reactions(self, user_id, channel_id, limit=None):
reactions = []
channel = self.bot.get_channel(channel_id)
async for message in channel.history(limit=limit):
if len(message.reactions) == 0:
continue
for reaction in message.reactions:
users = await reaction.users().flatten()
user_ids = map(lambda user: user.id, users)
if user_id in user_ids:
reactions.append(reaction)
return reactions
def create_log_message(self, emoji, action, user, channel, reason=""):
msg = (
f"{emoji} **{action}** \n"
f"from {self.bot.escape_message(user.name)} ({user.id}), in {channel.mention}"
)
if reason != "":
msg += f":\n`{reason}`"
return msg
async def clean_up_raw_text_file_message(self, message):
embeds = message.embeds
if len(embeds) == 0:
return
fields = embeds[0].fields
for field in fields:
if field.name == "Message ID":
files_channel = self.bot.get_channel(config.list_files_channel)
file_message = await files_channel.fetch_message(int(field.value))
await file_message.delete()
await message.edit(embed=None)
async def link_list_item(self, ctx, channel: discord.TextChannel, number: int):
if number <= 0:
await ctx.send(f"Number must be greater than 0.")
return False
if channel.id not in config.list_channels:
await ctx.send(f"{channel.mention} is not a list channel.")
return False
counter = 0
async for message in channel.history(limit=None, oldest_first=True):
if message.content.strip():
counter += 1
if counter == number:
embed = discord.Embed(
title=f"Item #{number} in #{channel.name}",
description=message.content.replace("โ€‹", "").strip(),
)
embed.add_field(name="Jump URL", value=f"[Jump!]({message.jump_url})")
await ctx.send(content="", embed=embed)
return True
await ctx.send(f"Unable to find item #{number} in {channel.mention}.")
return False
async def cache_message(self, message):
msg = {
"has_attachment": False,
"attachment_filename": "",
"attachment_data": b"",
"content": message.content,
}
if len(message.attachments) != 0:
attachment = next(
(
a
for a in message.attachments
if os.path.splitext(a.filename)[1] in [".png", ".jpg", ".jpeg"]
),
None,
)
if attachment is not None:
msg["has_attachment"] = True
msg["attachment_filename"] = attachment.filename
msg["attachment_data"] = await attachment.read()
return msg
async def send_cached_message(self, channel, message):
if message["has_attachment"] == True:
file = discord.File(
io.BytesIO(message["attachment_data"]),
filename=message["attachment_filename"],
)
await channel.send(content=message["content"], file=file)
else:
await channel.send(content=message["content"])
# Commands
@commands.guild_only()
@commands.check(check_if_verified)
@commands.command(aliases=["list"])
async def listitem(self, ctx, channel: discord.TextChannel, number: int):
"""Link to a specific list item."""
await self.link_list_item(ctx, channel, number)
@commands.guild_only()
@commands.check(check_if_verified)
@commands.command(aliases=["rule"])
async def rules(self, ctx, number: int):
"""Link to a specific list item in #rules"""
channel = ctx.guild.get_channel(config.rules_channel)
await self.link_list_item(ctx, channel, number)
@commands.guild_only()
@commands.check(check_if_staff)
@commands.command(aliases=["warnrule", "ruleswarn", "warnrules"])
async def rulewarn(
self, ctx, target: discord.Member, number: int, *, reason: str = ""
):
if "Mod" not in self.bot.cogs:
await ctx.send("Mod cog must be loaded to run this command.")
return
mod_cog = self.bot.cogs["Mod"]
warn_command = None
for command in mod_cog.get_commands():
if command.name == "warn":
warn_command = command
break
if warn_command is None:
await ctx.send("Unable to find the warn command from the Mod cog.")
return
if not await warn_command.can_run(ctx):
await ctx.send("Unable to run the warn command from the Mod cog.")
return
if (
len(warn_command.params) != 4
or "self" not in warn_command.params
or "ctx" not in warn_command.params
or "target" not in warn_command.params
or "reason" not in warn_command.params
):
await ctx.send("Warn's signature has changed please update the Lists cog.")
return
channel = ctx.guild.get_channel(config.rules_channel)
if await self.link_list_item(ctx, channel, number):
await warn_command.callback(
mod_cog, ctx=ctx, target=target, reason=f"Rule {number} - {reason}"
)
@commands.guild_only()
@commands.check(check_if_verified)
@commands.command(aliases=["faq"])
async def support(self, ctx, number: int):
"""Link to a specific list item in #support-faq"""
channel = ctx.guild.get_channel(config.support_faq_channel)
await self.link_list_item(ctx, channel, number)
@commands.guild_only()
@commands.check(check_if_verified)
@commands.command(aliases=["es", "fs", "acid", "piracy"])
async def patches(self, ctx):
"""Link to the list item in #support-faq about patches"""
channel = ctx.guild.get_channel(config.support_faq_channel)
await self.link_list_item(ctx, channel, 1)
# Listeners
@Cog.listener()
async def on_raw_reaction_add(self, payload):
await self.bot.wait_until_ready()
# We only care about reactions in Rules, and Support FAQ
if payload.channel_id not in config.list_channels:
return
channel = self.bot.get_channel(payload.channel_id)
message = await channel.fetch_message(payload.message_id)
member = channel.guild.get_member(payload.user_id)
user = self.bot.get_user(payload.user_id)
reaction = next(
(
reaction
for reaction in message.reactions
if str(reaction.emoji) == str(payload.emoji)
),
None,
)
if reaction is None:
return
# Only staff can add reactions in these channels.
if not self.check_if_target_is_staff(member):
await reaction.remove(user)
return
# Reactions are only allowed on messages from the bot.
if not message.author.bot:
await reaction.remove(user)
return
# Only certain reactions are allowed.
if not self.is_reaction_valid(reaction):
await reaction.remove(user)
return
# Remove all other reactions from user in this channel.
for r in await self.find_reactions(payload.user_id, payload.channel_id):
if r.message.id != message.id or (
r.message.id == message.id and str(r.emoji) != str(reaction.emoji)
):
await r.remove(user)
# When editing we want to provide the user a copy of the raw text.
if self.is_edit(reaction.emoji) and config.list_files_channel != 0:
files_channel = self.bot.get_channel(config.list_files_channel)
file = discord.File(
io.BytesIO(message.content.encode("utf-8")),
filename=f"{message.id}.txt",
)
file_message = await files_channel.send(file=file)
embed = discord.Embed(
title="Click here to get the raw text to modify.",
url=f"{file_message.attachments[0].url}?",
)
embed.add_field(name="Message ID", value=file_message.id, inline=False)
await message.edit(embed=embed)
@Cog.listener()
async def on_raw_reaction_remove(self, payload):
await self.bot.wait_until_ready()
# We only care about reactions in Rules, and Support FAQ
if payload.channel_id not in config.list_channels:
return
channel = self.bot.get_channel(payload.channel_id)
message = await channel.fetch_message(payload.message_id)
# Reaction was removed from a message we don't care about.
if not message.author.bot:
return
# We want to remove the embed we added.
if self.is_edit(payload.emoji) and config.list_files_channel != 0:
await self.clean_up_raw_text_file_message(message)
@Cog.listener()
async def on_message(self, message):
await self.bot.wait_until_ready()
# We only care about messages in Rules, and Support FAQ
if message.channel.id not in config.list_channels:
return
# We don't care about messages from bots.
if message.author.bot:
return
# Only staff can modify lists.
if not self.check_if_target_is_staff(message.author):
await message.delete()
return
log_channel = self.bot.get_channel(config.log_channel)
channel = message.channel
content = message.content
user = message.author
attachment_filename = None
attachment_data = None
if len(message.attachments) != 0:
# Lists will only reupload the first image.
attachment = next(
(
a
for a in message.attachments
if os.path.splitext(a.filename)[1] in [".png", ".jpg", ".jpeg"]
),
None,
)
if attachment is not None:
attachment_filename = attachment.filename
attachment_data = await attachment.read()
await message.delete()
reactions = await self.find_reactions(user.id, channel.id)
# Add to the end of the list if there is no reactions or somehow more
# than one.
if len(reactions) != 1:
if attachment_filename is not None and attachment_data is not None:
file = discord.File(
io.BytesIO(attachment_data), filename=attachment_filename
)
await channel.send(content=content, file=file)
else:
await channel.send(content)
for reaction in reactions:
await reaction.remove(user)
await log_channel.send(
self.create_log_message("๐Ÿ’ฌ", "List item added:", user, channel)
)
return
targeted_reaction = reactions[0]
targeted_message = targeted_reaction.message
if self.is_edit(targeted_reaction):
if config.list_files_channel != 0:
await self.clean_up_raw_text_file_message(targeted_message)
await targeted_message.edit(content=content)
await targeted_reaction.remove(user)
await log_channel.send(
self.create_log_message("๐Ÿ“", "List item edited:", user, channel)
)
elif self.is_delete(targeted_reaction):
await targeted_message.delete()
await log_channel.send(
self.create_log_message(
"โŒ", "List item deleted:", user, channel, content
)
)
elif self.is_recycle(targeted_reaction):
messages = [await self.cache_message(targeted_message)]
for message in await channel.history(
limit=None, after=targeted_message, oldest_first=True
).flatten():
messages.append(await self.cache_message(message))
await channel.purge(limit=len(messages) + 1, bulk=True)
for message in messages:
await self.send_cached_message(channel, message)
await log_channel.send(
self.create_log_message(
"โ™ป", "List item recycled:", user, channel, content
)
)
elif self.is_insert_above(targeted_reaction):
messages = [await self.cache_message(targeted_message)]
for message in await channel.history(
limit=None, after=targeted_message, oldest_first=True
).flatten():
messages.append(await self.cache_message(message))
await channel.purge(limit=len(messages) + 1, bulk=True)
if attachment_filename is not None and attachment_data is not None:
file = discord.File(
io.BytesIO(attachment_data), filename=attachment_filename
)
await channel.send(content=content, file=file)
else:
await channel.send(content)
for message in messages:
await self.send_cached_message(channel, message)
await log_channel.send(
self.create_log_message("๐Ÿ’ฌ", "List item added: