Browse Source

Robocronp: Add robocronp and some timed commands

- Added: .timemute
- Added: .timeban
- Added: .listjobs
- Added: .deletejob
- Many bugfixes are also included~

You'll need to restart your copy of robocop-ng after pulling this.
main
Ave Ozkal 2 years ago
parent
commit
b552888f1b
9 changed files with 311 additions and 21 deletions
  1. +23
    -5
      README.md
  2. +13
    -8
      Robocop.py
  3. +20
    -5
      cogs/lockdown.py
  4. +2
    -1
      cogs/mod.py
  5. +1
    -1
      cogs/mod_note.py
  6. +122
    -0
      cogs/mod_timed.py
  7. +1
    -1
      cogs/mod_userlog.py
  8. +92
    -0
      cogs/robocronp.py
  9. +37
    -0
      helpers/robocronp.py

+ 23
- 5
README.md View File

@ -1,6 +1,6 @@
# Robocop-ng
Next-gen rewrite of Kurisu/Robocop bot used on ReSwitched bot with discord.py rewrite, designed to be clean, fast and un-bloated.
Next-gen rewrite of Kurisu/Robocop bot used on ReSwitched bot with discord.py rewrite, designed to be relatively clean, consistent and un-bloated.
Code is based on https://gitlab.com/ao/dpybotbase and https://github.com/916253/Kurisu-Reswitched.
@ -31,7 +31,7 @@ If you're moving from Kurisu/Robocop, and want to preserve your data, you'll wan
## TODO
All Kurisu/Robocop features are now supported.
All Robocop features are now supported.
<details>
<summary>List of added Kurisu/Robocop features</summary>
@ -83,9 +83,14 @@ Main goal of this project is to get Robocop functionality done, secondary goal i
- [ ] New feature: Modmail
- [ ] New feature: Submiterr (relies on modmail)
- [ ] New feature: Highlights (problematic words automatically get posted to modmail channel, relies on modmail)
- [ ] A system for running tasks in background with an interval (will be called robocronp)
- [ ] New moderation feature: mutetime (mute with time, relies on robocronp)
- [ ] Feature creep: Shortlink completion (gl/ao/etc)
- [ ] Feature creep: Pleroma embedding
- [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)
- [ ] New moderation feature: timelock (channel lockdown with time, 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!)
@ -93,12 +98,25 @@ Main goal of this project is to get Robocop functionality done, secondary goal i
- [x] New self-moderation feature: .mywarns
- [x] Remove sh, remove risky stuff from eval
<details>
<summary>TODO for robocronp</summary>
<p>
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>
---
## Thanks to
- ReSwitched community, for being amazing
- ihaveamac and f916253 for the original kurisu/robocop
- ihaveamac/ihaveahax and f916253 for the original kurisu/robocop
- tomGER for working hard on rewriting the .err/.serr commands, those were a nightmare
- misson20000 for adding in reaction removal feature and putting up with my many BS requests on PR reviews

+ 13
- 8
Robocop.py View File

@ -1,4 +1,5 @@
import os
import asyncio
import sys
import logging
import logging.handlers
@ -38,27 +39,31 @@ def get_prefix(bot, message):
wanted_jsons = ["data/restrictions.json",
"data/robocronptab.json",
"data/userlog.json"]
initial_extensions = ['cogs.common',
'cogs.admin',
'cogs.basic',
'cogs.err',
'cogs.verification',
'cogs.logs',
'cogs.lockdown',
'cogs.legacy',
'cogs.links',
'cogs.mod',
'cogs.mod_note',
'cogs.mod_reacts',
'cogs.mod_userlog',
'cogs.mod_timed',
'cogs.basic',
'cogs.logs',
'cogs.err',
'cogs.lockdown',
'cogs.legacy',
'cogs.links',
'cogs.robocronp',
'cogs.meme']
bot = commands.Bot(command_prefix=get_prefix,
description=config.bot_description, pm_help=True)
bot.log = log
bot.loop = asyncio.get_event_loop()
bot.config = config
bot.script_name = script_name
bot.wanted_jsons = wanted_jsons
@ -68,7 +73,7 @@ if __name__ == '__main__':
try:
bot.load_extension(extension)
except Exception as e:
log.error(f'Failed to load extension {extension}.', file=sys.stderr)
log.error(f'Failed to load extension {extension}.')
log.error(traceback.print_exc())
@ -181,4 +186,4 @@ for wanted_json in wanted_jsons:
with open(wanted_json, "w") as f:
f.write("{}")
bot.run(config.token, bot=True, reconnect=True)
bot.run(config.token, bot=True, reconnect=True, loop=bot.loop)

+ 20
- 5
cogs/lockdown.py View File

@ -8,17 +8,28 @@ class Lockdown:
def __init__(self, bot):
self.bot = bot
async def unlock_for_staff(self, channel: discord.TextChannel, issuer):
for role in config.staff_role_ids:
try:
await channel.set_permissions(channel.guild.get_role(role),
send_messages=False,
reason=str(issuer))
except:
pass
@commands.guild_only()
@commands.check(check_if_staff)
@commands.command()
async def lock(self, ctx, channel: discord.TextChannel = None,
soft: bool = False):
"""Prevents people from speaking in current channel, staff only."""
"""Prevents people from speaking in a channel, staff only.
Defaults to current channel."""
if not channel:
channel = ctx.channel
log_channel = self.bot.get_channel(config.log_channel)
if ctx.channel.id in config.community_channels:
if channel.id in config.community_channels:
roles = [config.named_roles["community"],
config.named_roles["hacker"]]
else:
@ -26,9 +37,11 @@ class Lockdown:
ctx.guild.default_role.id]
for role in roles:
await ctx.channel.set_permissions(ctx.guild.get_role(role),
send_messages=False,
reason=str(ctx.author))
await channel.set_permissions(channel.guild.get_role(role),
send_messages=False,
reason=str(ctx.author))
await self.unlock_for_staff(channel, ctx.author)
public_msg = "πŸ”’ Channel locked down. "
if not soft:
@ -57,6 +70,8 @@ class Lockdown:
roles = [config.named_roles["participant"],
ctx.guild.default_role.id]
await self.unlock_for_staff(channel, ctx.author)
for role in roles:
await ctx.channel.set_permissions(ctx.guild.get_role(role),
send_messages=True,


+ 2
- 1
cogs/mod.py View File

@ -116,7 +116,8 @@ class Mod:
chan_message += f"✏️ __Reason__: \"{reason}\""
else:
chan_message += "Please add an explanation below. In the future"\
", it is recommended to use `.ban <user> [reason]`"\
", it is recommended to use "\
"`.kick <user> [reason]`"\
" as the reason is automatically sent to the user."
log_channel = self.bot.get_channel(config.log_channel)


+ 1
- 1
cogs/mod_note.py View File

@ -15,7 +15,7 @@ class ModNote:
"""Adds a note to a user, staff only."""
userlog(target.id, ctx.author, note,
"notes", target.name)
await ctx.send(f"{target.mention}: noted!")
await ctx.send(f"{ctx.author.mention}: noted!")
@commands.guild_only()
@commands.check(check_if_staff)


+ 122
- 0
cogs/mod_timed.py View File

@ -0,0 +1,122 @@
import discord
import config
import time
from discord.ext import commands
from helpers.checks import check_if_staff
from helpers.robocronp import add_job
from helpers.userlogs import userlog
from helpers.restrictions import add_restriction
class ModTimed:
def __init__(self, bot):
self.bot = bot
def check_if_target_is_staff(self, target):
return any(r.id in config.staff_role_ids for r in target.roles)
@commands.guild_only()
@commands.bot_has_permissions(ban_members=True)
@commands.check(check_if_staff)
@commands.command()
async def timeban(self, ctx, target: discord.Member,
hours: int, *, reason: str = ""):
"""Bans a user for a specified amount of hours, staff only."""
# Hedge-proofing the code
if target == ctx.author:
return await ctx.send("You can't do mod actions on yourself.")
elif self.check_if_target_is_staff(target):
return await ctx.send("I can't ban this user as "
"they're a member of staff.")
userlog(target.id, ctx.author, f"{reason} (Timed, for {hours}h)",
"bans", target.name)
safe_name = self.bot.escape_message(str(target))
dm_message = f"You were banned from {ctx.guild.name}."
if reason:
dm_message += f" The given reason is: \"{reason}\"."
dm_message += f"\n\nThis ban will expire in {hours} hours."
try:
await target.send(dm_message)
except discord.errors.Forbidden:
# Prevents ban issues in cases where user blocked bot
# or has DMs disabled
pass
await target.ban(reason=f"{ctx.author}, reason: {reason}",
delete_message_days=0)
chan_message = f"β›” **Timed Ban**: {ctx.author.mention} banned "\
f"{target.mention} for {hours} hours | {safe_name}\n"\
f"🏷 __User ID__: {target.id}\n"
if reason:
chan_message += f"✏️ __Reason__: \"{reason}\""
else:
chan_message += "Please add an explanation below. In the future"\
", it is recommended to use `.ban <user> [reason]`"\
" as the reason is automatically sent to the user."
expiry_timestamp = time.time() + (hours * 3600)
add_job("unban", target.id, {"guild": ctx.guild.id}, expiry_timestamp)
log_channel = self.bot.get_channel(config.log_channel)
await log_channel.send(chan_message)
await ctx.send(f"{safe_name} is now b& for {hours} hours. πŸ‘")
@commands.guild_only()
@commands.check(check_if_staff)
@commands.command()
async def timemute(self, ctx, target: discord.Member,
hours: int, *, reason: str = ""):
"""Mutes a user for a specified amount of hours, staff only."""
# Hedge-proofing the code
if target == ctx.author:
return await ctx.send("You can't do mod actions on yourself.")
elif self.check_if_target_is_staff(target):
return await ctx.send("I can't mute this user as "
"they're a member of staff.")
userlog(target.id, ctx.author, f"{reason} (Timed, for {hours}h)",
"mutes", target.name)
safe_name = self.bot.escape_message(str(target))
dm_message = f"You were muted!"
if reason:
dm_message += f" The given reason is: \"{reason}\"."
dm_message += f"\n\nThis mute will expire in {hours} hours."
try:
await target.send(dm_message)
except discord.errors.Forbidden:
# Prevents kick issues in cases where user blocked bot
# or has DMs disabled
pass
mute_role = ctx.guild.get_role(config.mute_role)
await target.add_roles(mute_role, reason=str(ctx.author))
chan_message = f"πŸ”‡ **Timed Mute**: {ctx.author.mention} muted "\
f"{target.mention} for {hours} hours | {safe_name}\n"\
f"🏷 __User ID__: {target.id}\n"
if reason:
chan_message += f"✏️ __Reason__: \"{reason}\""
else:
chan_message += "Please add an explanation below. In the future, "\
"it is recommended to use `.mute <user> [reason]`"\
" as the reason is automatically sent to the user."
expiry_timestamp = time.time() + (hours * 3600)
add_job("unmute", target.id, {"guild": ctx.guild.id}, expiry_timestamp)
log_channel = self.bot.get_channel(config.log_channel)
await log_channel.send(chan_message)
await ctx.send(f"{target.mention} can no longer speak.")
add_restriction(target.id, config.mute_role)
def setup(bot):
bot.add_cog(ModTimed(bot))

+ 1
- 1
cogs/mod_userlog.py View File

@ -69,7 +69,7 @@ class ModUserlog:
if idx < 1:
return "Index is below 1!"
event = userlog[uid][event_type][idx - 1]
event_name = userlog_event_types[event_type].lower()
event_name = userlog_event_types[event_type]
embed = discord.Embed(color=discord.Color.dark_red(),
title=f"{event_name} {idx} on "
f"{event['timestamp']}",


+ 92
- 0
cogs/robocronp.py View File

@ -0,0 +1,92 @@
import asyncio
import config
import time
import discord
from discord.ext import commands
from helpers.robocronp import get_crontab, delete_job
from helpers.restrictions import remove_restriction
from helpers.checks import check_if_staff
class Robocronp:
def __init__(self, bot):
self.bot = bot
bot.loop.create_task(self.minutely())
# bot.loop.create_task(self.hourly())
@commands.guild_only()
@commands.check(check_if_staff)
@commands.command()
async def listjobs(self, ctx):
"""Lists timed robocronp jobs, staff only."""
ctab = get_crontab()
embed = discord.Embed(title=f"Active robocronp jobs")
for jobtype in ctab:
for jobtimestamp in ctab[jobtype]:
for job_name in ctab[jobtype][jobtimestamp]:
job_details = repr(ctab[jobtype][jobtimestamp][job_name])
embed.add_field(name=f"{jobtype} for {job_name}",
value=f"Timestamp: {jobtimestamp}, "
f"Details: {job_details}",
inline=False)
await ctx.send(embed=embed)
@commands.guild_only()
@commands.check(check_if_staff)
@commands.command(aliases=["removejob"])
async def deletejob(self, ctx, timestamp: str,
job_type: str, job_name: str):
"""Removes a timed robocronp job, staff only.
You'll need to supply:
- timestamp (like 1545981602)
- job type (like "unban")
- job name (userid, like 420332322307571713)
You can get all 3 from listjobs command."""
delete_job(timestamp, job_type, job_name)
await ctx.send(f"{ctx.author.mention}: Deleted!")
async def do_jobs(self, ctab, jobtype, timestamp):
for job_name in ctab[jobtype][timestamp]:
job_details = ctab[jobtype][timestamp][job_name]
if jobtype == "unban":
target_user = await self.bot.get_user_info(job_name)
target_guild = self.bot.get_guild(job_details["guild"])
await target_guild.unban(target_user,
reason="Robocronp: Timed ban expired.")
delete_job(timestamp, jobtype, job_name)
elif jobtype == "unmute":
remove_restriction(job_name, config.mute_role)
target_guild = self.bot.get_guild(job_details["guild"])
target_member = target_guild.get_member(int(job_name))
target_role = target_guild.get_role(config.mute_role)
await target_member.remove_roles(target_role,
reason="Robocronp: Timed "
"mute expired.")
delete_job(timestamp, jobtype, job_name)
async def minutely(self):
await self.bot.wait_until_ready()
while not self.bot.is_closed():
try:
ctab = get_crontab()
timestamp = time.time()
for jobtype in ctab:
for jobtimestamp in ctab[jobtype]:
if timestamp > int(jobtimestamp):
await self.do_jobs(ctab, jobtype, jobtimestamp)
except:
# Don't kill cronjobs if something goes wrong.
pass
await asyncio.sleep(60)
# async def hourly(self):
# await self.bot.wait_until_ready()
# while not self.bot.is_closed():
# # Your stuff goes here
# await asyncio.sleep(3600)
def setup(bot):
bot.add_cog(Robocronp(bot))

+ 37
- 0
helpers/robocronp.py View File

@ -0,0 +1,37 @@
import json
import math
def get_crontab():
with open("data/robocronptab.json", "r") as f:
return json.load(f)
def set_crontab(contents):
with open("data/robocronptab.json", "w") as f:
f.write(contents)
def add_job(job_type, job_name, job_details, timestamp):
timestamp = str(math.floor(timestamp))
job_name = str(job_name)
ctab = get_crontab()
if job_type not in ctab:
ctab[job_type] = {}
if timestamp not in ctab[job_type]:
ctab[job_type][timestamp] = {}
ctab[job_type][timestamp][job_name] = job_details
set_crontab(json.dumps(ctab))
def delete_job(timestamp, job_type, job_name):
timestamp = str(timestamp)
job_name = str(job_name)
ctab = get_crontab()
del ctab[job_type][timestamp][job_name]
set_crontab(json.dumps(ctab))

Loading…
Cancel
Save