jarvis/jarvis-bot
jarvis/jarvis-bot
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

V2.0 Beta 2

Browse source 
Closes #138 and #128
This commit is contained in:
Zeva Rose 2022-05-05 22:01:25 +00:00
commit a7efedd46a
38 changed files with 2584 additions and 734 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

67
jarvis/__init__.py
View file

@ -1,42 +1,53 @@
"""Main JARVIS package."""
import logging
from importlib.metadata import version as _v
from dis_snek import Intents
import aioredis
import jurigged
import rook
from jarvis_core.db import connect
from jarvis_core.log import get_logger
from naff import Intents
from jarvis import utils
from jarvis import const
from jarvis.client import Jarvis
from jarvis.cogs import __path__ as cogs_path
from jarvis.config import JarvisConfig
from jarvis.utils import get_extensions
try:
__version__ = _v("jarvis")
except Exception:
__version__ = "0.0.0"
jconfig = JarvisConfig.from_yaml()
logger = get_logger("jarvis")
logger.setLevel(jconfig.log_level)
file_handler = logging.FileHandler(filename="jarvis.log", encoding="UTF-8", mode="w")
file_handler.setFormatter(
logging.Formatter("[%(asctime)s] [%(name)s] [%(levelname)8s] %(message)s")
)
logger.addHandler(file_handler)
intents = Intents.DEFAULT | Intents.MESSAGES | Intents.GUILD_MEMBERS | Intents.GUILD_MESSAGES
restart_ctx = None
jarvis = Jarvis(
intents=intents,
sync_interactions=jconfig.sync,
delete_unused_application_cmds=True,
send_command_tracebacks=False,
)
__version__ = const.__version__
async def run() -> None:
"""Run JARVIS"""
jconfig = JarvisConfig.from_yaml()
logger = get_logger("jarvis", show_locals=jconfig.log_level == "DEBUG")
logger.setLevel(jconfig.log_level)
file_handler = logging.FileHandler(filename="jarvis.log", encoding="UTF-8", mode="w")
file_handler.setFormatter(
logging.Formatter("[%(asctime)s] [%(name)s] [%(levelname)8s] %(message)s")
)
logger.addHandler(file_handler)
intents = (
Intents.DEFAULT | Intents.MESSAGES | Intents.GUILD_MEMBERS | Intents.GUILD_MESSAGE_CONTENT
)
redis_config = jconfig.redis.copy()
redis_host = redis_config.pop("host")
redis = await aioredis.from_url(redis_host, decode_responses=True, **redis_config)
jarvis = Jarvis(
intents=intents,
sync_interactions=jconfig.sync,
delete_unused_application_cmds=True,
send_command_tracebacks=False,
redis=redis,
)
if jconfig.log_level == "DEBUG":
jurigged.watch()
if jconfig.rook_token:
rook.start(token=jconfig.rook_token, labels={"env": "dev"})
logger.info("Starting JARVIS")
logger.debug("Connecting to database")
connect(**jconfig.mongo["connect"], testing=jconfig.mongo["database"] != "jarvis")
@ -44,9 +55,9 @@ async def run() -> None:
# jconfig.get_db_config()
logger.debug("Loading extensions")
for extension in utils.get_extensions():
for extension in get_extensions(cogs_path):
jarvis.load_extension(extension)
logger.debug(f"Loaded {extension}")
logger.debug("Loaded %s", extension)
jarvis.max_messages = jconfig.max_messages
logger.debug("Running JARVIS")

328
jarvis/client.py
View file

@ -1,34 +1,48 @@
"""Custom JARVIS client."""
import asyncio
import logging
import re
import traceback
from datetime import datetime, timezone
from datetime import datetime, timedelta, timezone
from aiohttp import ClientSession
from dis_snek import Snake, listen
from dis_snek.api.events.discord import (
from jarvis_core.db import q
from jarvis_core.db.models import (
Action,
Autopurge,
Autoreact,
Modlog,
Note,
Roleping,
Setting,
Warning,
)
from jarvis_core.filters import invites, url
from jarvis_core.util.ansi import RESET, Fore, Format, fmt
from naff import Client, listen
from naff.api.events.discord import (
MemberAdd,
MemberRemove,
MemberUpdate,
MessageCreate,
MessageDelete,
MessageUpdate,
)
from dis_snek.client.errors import CommandCheckFailure, CommandOnCooldown
from dis_snek.client.utils.misc_utils import find_all
from dis_snek.models.discord.channel import DMChannel
from dis_snek.models.discord.embed import EmbedField
from dis_snek.models.discord.enums import Permissions
from dis_snek.models.discord.message import Message
from dis_snek.models.snek.context import Context, InteractionContext, MessageContext
from dis_snek.models.snek.tasks.task import Task
from dis_snek.models.snek.tasks.triggers import IntervalTrigger
from jarvis_core.db import q
from jarvis_core.db.models import Autopurge, Autoreact, Roleping, Setting, Warning
from jarvis_core.filters import invites, url
from jarvis_core.util import build_embed
from jarvis_core.util.ansi import RESET, Fore, Format, fmt
from naff.api.events.internal import Button
from naff.client.errors import CommandCheckFailure, CommandOnCooldown, HTTPException
from naff.client.utils.misc_utils import find_all, get
from naff.models.discord.channel import DMChannel
from naff.models.discord.embed import Embed, EmbedField
from naff.models.discord.enums import AuditLogEventType, Permissions
from naff.models.discord.message import Message
from naff.models.discord.user import Member
from naff.models.naff.context import Context, InteractionContext, PrefixedContext
from naff.models.naff.tasks.task import Task
from naff.models.naff.tasks.triggers import IntervalTrigger
from pastypy import AsyncPaste as Paste
from jarvis import const
from jarvis.utils import build_embed
from jarvis.utils.embeds import warning_embed
DEFAULT_GUILD = 862402786116763668
@ -54,9 +68,11 @@ VAL_FMT = fmt(Fore.WHITE)
CMD_FMT = fmt(Fore.GREEN, Format.BOLD)
class Jarvis(Snake):
class Jarvis(Client):
def __init__(self, *args, **kwargs): # noqa: ANN002 ANN003
redis = kwargs.pop("redis")
super().__init__(*args, **kwargs)
self.redis = redis
self.logger = logging.getLogger(__name__)
self.phishing_domains = []
self.pre_run_callback = self._prerun
@ -71,19 +87,29 @@ class Jarvis(Snake):
self.logger.debug(f"Found {len(data)} changes to phishing domains")
add = 0
sub = 0
for update in data:
if update["type"] == "add":
if update["domain"] not in self.phishing_domains:
self.phishing_domains.append(update["domain"])
for domain in update["domains"]:
if domain not in self.phishing_domains:
add += 1
self.phishing_domains.append(domain)
elif update["type"] == "delete":
if update["domain"] in self.phishing_domains:
self.phishing_domains.remove(update["domain"])
for domain in update["domains"]:
if domain in self.phishing_domains:
sub -= 1
self.phishing_domains.remove(domain)
self.logger.debug(f"{add} additions, {sub} removals")
async def _prerun(self, ctx: Context, *args, **kwargs) -> None:
name = ctx.invoked_name
name = ctx.invoke_target
if isinstance(ctx, InteractionContext) and ctx.target_id:
kwargs["context target"] = ctx.target
args = " ".join(f"{k}:{v}" for k, v in kwargs.items())
elif isinstance(ctx, PrefixedContext):
args = " ".join(args)
self.logger.debug(f"Running command `{name}` with args: {args or 'None'}")
async def _sync_domains(self) -> None:
@ -96,21 +122,35 @@ class Jarvis(Snake):
@listen()
async def on_ready(self) -> None:
"""Lepton on_ready override."""
"""NAFF on_ready override."""
try:
await self._sync_domains()
self._update_domains.start()
except Exception as e:
self.logger.error("Failed to load anti-phishing", exc_info=e)
self.logger.info("Logged in as {}".format(self.user)) # noqa: T001
self.logger.info("Connected to {} guild(s)".format(len(self.guilds))) # noqa: T001
self.logger.info("Current version: {}".format(const.__version__))
self.logger.info( # noqa: T001
"https://discord.com/api/oauth2/authorize?client_id="
"{}&permissions=8&scope=bot%20applications.commands".format(self.user.id)
)
async def on_error(self, source: str, error: Exception, *args, **kwargs) -> None:
"""NAFF on_error override."""
if isinstance(error, HTTPException):
errors = error.search_for_message(error.errors)
out = f"HTTPException: {error.status}|{error.response.reason}: " + "\n".join(errors)
self.logger.error(out, exc_info=error)
else:
self.logger.error(f"Ignoring exception in {source}", exc_info=error)
async def on_command_error(
self, ctx: Context, error: Exception, *args: list, **kwargs: dict
) -> None:
"""Lepton on_command_error override."""
self.logger.debug(f"Handling error in {ctx.invoked_name}: {error}")
"""NAFF on_command_error override."""
name = ctx.invoke_target
self.logger.debug(f"Handling error in {name}: {error}")
if isinstance(error, CommandOnCooldown):
await ctx.send(str(error), ephemeral=True)
return
@ -131,7 +171,7 @@ class Jarvis(Snake):
if isinstance(v, str) and len(v) > 100:
v = v[97] + "..."
arg_str += f"{v}\n"
elif isinstance(ctx, MessageContext):
elif isinstance(ctx, PrefixedContext):
for v in ctx.args:
if isinstance(v, str) and len(v) > 100:
v = v[97] + "..."
@ -143,11 +183,15 @@ class Jarvis(Snake):
full_message = ERROR_MSG.format(
guild_name=ctx.guild.name,
error_time=error_time,
invoked_name=ctx.invoked_name,
invoked_name=name,
arg_str=arg_str,
callback_args=callback_args,
callback_kwargs=callback_kwargs,
)
tb = traceback.format_exception(error)
if isinstance(error, HTTPException):
errors = error.search_for_message(error.errors)
tb[-1] = f"HTTPException: {error.status}|{error.response.reason}: " + "\n".join(errors)
error_message = "".join(traceback.format_exception(error))
if len(full_message + error_message) >= 1800:
error_message = "\n ".join(error_message.split("\n"))
@ -167,12 +211,16 @@ class Jarvis(Snake):
f"\nException:\n```py\n{error_message}\n```"
)
await ctx.send("Whoops! Encountered an error. The error has been logged.", ephemeral=True)
try:
return await super().on_command_error(ctx, error, *args, **kwargs)
except Exception as e:
self.logger.error("Uncaught exception", exc_info=e)
# Modlog
async def on_command(self, ctx: Context) -> None:
"""Lepton on_command override."""
if not isinstance(ctx.channel, DMChannel) and ctx.invoked_name not in ["pw"]:
"""NAFF on_command override."""
name = ctx.invoke_target
if not isinstance(ctx.channel, DMChannel) and name not in ["pw"]:
modlog = await Setting.find_one(q(guild=ctx.guild.id, setting="activitylog"))
if modlog:
channel = await ctx.guild.fetch_channel(modlog.value)
@ -186,7 +234,7 @@ class Jarvis(Snake):
if len(v) > 100:
v = v[:97] + "..."
args.append(f"{KEY_FMT}{k}:{VAL_FMT}{v}{RESET}")
elif isinstance(ctx, MessageContext):
elif isinstance(ctx, PrefixedContext):
for v in ctx.args:
if isinstance(v, str) and len(v) > 100:
v = v[97] + "..."
@ -195,7 +243,7 @@ class Jarvis(Snake):
fields = [
EmbedField(
name="Command",
value=f"```ansi\n{CMD_FMT}{ctx.invoked_name}{RESET} {args}\n```",
value=f"```ansi\n{CMD_FMT}{ctx.invoke_target}{RESET} {args}\n```",
inline=False,
),
]
@ -242,13 +290,131 @@ class Jarvis(Snake):
channel = await guild.fetch_channel(log.channel)
embed = build_embed(
title="Member Left",
desciption=f"{user.username}#{user.discriminator} left {guild.name}",
description=f"{user.username}#{user.discriminator} left {guild.name}",
fields=[],
)
embed.set_author(name=user.username, icon_url=user.avatar.url)
embed.set_footer(text=f"{user.username}#{user.discriminator} | {user.id}")
await channel.send(embed=embed)
async def process_verify(self, before: Member, after: Member) -> Embed:
"""Process user verification."""
auditlog = await after.guild.fetch_audit_log(
user_id=before.id, action_type=AuditLogEventType.MEMBER_ROLE_UPDATE
)
audit_event = get(auditlog.events, reason="Verification passed")
if audit_event:
admin_mention = "[N/A]"
admin_text = "[N/A]"
if admin := await after.guild.fet_member(audit_event.user_id):
admin_mention = admin.mention
admin_text = f"{admin.username}#{admin.discriminator}"
fields = (
EmbedField(name="Moderator", value=f"{admin_mention} ({admin_text})"),
EmbedField(name="Reason", value=audit_event.reason),
)
embed = build_embed(
title="User Verified",
description=f"{after.mention} was verified",
fields=fields,
color="#fc9e3f",
)
embed.set_author(name=after.display_name, icon_url=after.display_avatar.url)
embed.set_footer(text=f"{after.username}#{after.discriminator} | {after.id}")
return embed
async def process_rolechange(self, before: Member, after: Member) -> Embed:
"""Process role changes."""
if before.roles == after.roles:
return
new_roles = []
removed_roles = []
for role in before.roles:
if role not in after.roles:
removed_roles.append(role)
for role in after.roles:
if role not in before.roles:
new_roles.append(role)
new_text = "\n".join(role.mention for role in new_roles) or "None"
removed_text = "\n".join(role.mention for role in removed_roles) or "None"
fields = (
EmbedField(name="Added Roles", value=new_text),
EmbedField(name="Removed Roles", value=removed_text),
)
embed = build_embed(
title="User Roles Changed",
description=f"{after.mention} had roles changed",
fields=fields,
color="#fc9e3f",
)
embed.set_author(name=after.display_name, icon_url=after.display_avatar.url)
embed.set_footer(text=f"{after.username}#{after.discriminator} | {after.id}")
return embed
async def process_rename(self, before: Member, after: Member) -> None:
"""Process name change."""
if (
before.nickname == after.nickname
and before.discriminator == after.discriminator
and before.username == after.username
):
return
fields = (
EmbedField(
name="Before",
value=f"{before.display_name} ({before.username}#{before.discriminator})",
),
EmbedField(
name="After", value=f"{after.display_name} ({after.username}#{after.discriminator})"
),
)
embed = build_embed(
title="User Renamed",
description=f"{after.mention} changed their name",
fields=fields,
color="#fc9e3f",
)
embed.set_author(name=after.display_name, icon_url=after.display_avatar.url)
embed.set_footer(text=f"{after.username}#{after.discriminator} | {after.id}")
return embed
@listen()
async def on_member_update(self, event: MemberUpdate) -> None:
"""Handle on_member_update event."""
before = event.before
after = event.after
if (before.display_name == after.display_name and before.roles == after.roles) or (
not after or not before
):
return
log = await Setting.find_one(q(guild=before.guild.id, setting="activitylog"))
if log:
channel = await before.guild.fetch_channel(log.value)
await asyncio.sleep(0.5) # Wait for audit log
embed = None
if before._role_ids != after._role_ids:
verified = await Setting.find_one(q(guild=before.guild.id, setting="verified"))
v_role = None
if verified:
v_role = await before.guild.fetch_role(verified.value)
if not v_role:
self.logger.debug(f"Guild {before.guild.id} verified role no longer exists")
await verified.delete()
else:
if not before.has_role(v_role) and after.has_role(v_role):
embed = await self.process_verify(before, after)
embed = embed or await self.process_rolechange(before, after)
embed = embed or await self.process_rename(before, after)
if embed:
await channel.send(embed=embed)
# Message
async def autopurge(self, message: Message) -> None:
"""Handle autopurge events."""
@ -308,10 +474,13 @@ class Jarvis(Snake):
await message.delete()
except Exception:
self.logger.debug("Message deleted before action taken")
expires_at = datetime.now(tz=timezone.utc) + timedelta(hours=24)
await Warning(
active=True,
admin=self.user.id,
duration=24,
expires_at=expires_at,
guild=message.guild.id,
reason="Sent an invite link",
user=message.author.id,
@ -338,10 +507,12 @@ class Jarvis(Snake):
self.logger.debug(
f"Massmention threshold on {message.guild.id}/{message.channel.id}/{message.id}"
)
expires_at = datetime.now(tz=timezone.utc) + timedelta(hours=24)
await Warning(
active=True,
admin=self.user.id,
duration=24,
expires_at=expires_at,
guild=message.guild.id,
reason="Mass Mention",
user=message.author.id,
@ -351,19 +522,19 @@ class Jarvis(Snake):
async def roleping(self, message: Message) -> None:
"""Handle roleping events."""
try:
if message.author.has_permission(Permissions.MANAGE_GUILD):
return
except Exception as e:
self.logger.error("Failed to get permissions, pretending check failed", exc_info=e)
if await Roleping.collection.count_documents(q(guild=message.guild.id, active=True)) == 0:
return
rolepings = await Roleping.find(q(guild=message.guild.id, active=True)).to_list(None)
# Get all role IDs involved with message
roles = []
async for mention in message.mention_roles:
roles.append(mention.id)
roles = [x.id async for x in message.mention_roles]
async for mention in message.mention_users:
for role in mention.roles:
roles.append(role.id)
roles += [x.id for x in mention.roles]
if not roles:
return
@ -381,12 +552,15 @@ class Jarvis(Snake):
user_is_admin = message.author.has_permission(Permissions.ADMINISTRATOR)
# Check if user in a bypass list
def check_has_role(roleping: Roleping) -> bool:
return any(role.id in roleping.bypass["roles"] for role in message.author.roles)
user_has_bypass = False
for roleping in rolepings:
if message.author.id in roleping.bypass["users"]:
user_has_bypass = True
break
if any(role.id in roleping.bypass["roles"] for role in message.author.roles):
if check_has_role(roleping):
user_has_bypass = True
break
@ -394,10 +568,12 @@ class Jarvis(Snake):
self.logger.debug(
f"Rolepinged role in {message.guild.id}/{message.channel.id}/{message.id}"
)
expires_at = datetime.now(tz=timezone.utc) + timedelta(hours=24)
await Warning(
active=True,
admin=self.user.id,
duration=24,
expires_at=expires_at,
guild=message.guild.id,
reason="Pinged a blocked role/user with a blocked role",
user=message.author.id,
@ -412,10 +588,12 @@ class Jarvis(Snake):
self.logger.debug(
f"Phishing url `{m}` detected in {message.guild.id}/{message.channel.id}/{message.id}"
)
expires_at = datetime.now(tz=timezone.utc) + timedelta(hours=24)
await Warning(
active=True,
admin=self.user.id,
duration=24,
expires_at=expires_at,
guild=message.guild.id,
reason="Phishing URL",
user=message.author.id,
@ -441,10 +619,12 @@ class Jarvis(Snake):
self.logger.debug(
f"Scam url `{match.string}` detected in {message.guild.id}/{message.channel.id}/{message.id}"
)
expires_at = datetime.now(tz=timezone.utc) + timedelta(hours=24)
await Warning(
active=True,
admin=self.user.id,
duration=24,
expires_at=expires_at,
guild=message.guild.id,
reason="Unsafe URL",
user=message.author.id,
@ -511,7 +691,7 @@ class Jarvis(Snake):
)
await channel.send(embed=embed)
except Exception as e:
self.logger.warn(
self.logger.warning(
f"Failed to process edit {before.guild.id}/{before.channel.id}/{before.id}: {e}"
)
if not isinstance(after.channel, DMChannel) and not after.author.bot:
@ -587,6 +767,72 @@ class Jarvis(Snake):
)
await channel.send(embed=embed)
except Exception as e:
self.logger.warn(
self.logger.warning(
f"Failed to process edit {message.guild.id}/{message.channel.id}/{message.id}: {e}"
)
@listen()
async def on_button(self, event: Button) -> None:
"""Process button events."""
context = event.context
if not context.deferred and not context.responded:
await context.defer(ephemeral=True)
if not context.custom_id.startswith("modcase|"):
return await super().on_button(event)
if not context.author.has_permission(Permissions.MODERATE_MEMBERS):
return
user_key = f"msg|{context.message.id}"
action_key = ""
if context.custom_id == "modcase|yes":
if user_id := await self.redis.get(user_key):
action_key = f"{user_id}|{context.guild.id}"
if (user := await context.guild.fetch_member(user_id)) and (
action_data := await self.redis.get(action_key)
):
name, parent = action_data.split("|")[:2]
action = Action(action_type=name, parent=parent)
note = Note(
admin=context.author.id, content="Moderation case opened via message"
)
modlog = Modlog(
user=user.id,
admin=context.author.id,
guild=context.guild.id,
actions=[action],
notes=[note],
)
await modlog.commit()
fields = (
EmbedField(name="Admin", value=context.author.mention),
EmbedField(name="Opening Action", value=f"{name} {parent}"),
)
embed = build_embed(
title="Moderation Case Opened",
description=f"Moderation case opened against {user.mention}",
fields=fields,
)
embed.set_author(
name=user.username + "#" + user.discriminator,
icon_url=user.display_avatar.url,
)
await context.message.edit(embed=embed)
elif not user:
self.logger.debug("User no longer in guild")
await context.send("User no longer in guild", ephemeral=True)
else:
self.logger.warn("Unable to get action data ( %s )", action_key)
await context.send("Unable to get action data", ephemeral=True)
for row in context.message.components:
for component in row.components:
component.disabled = True
await context.message.edit(components=context.message.components)
msg = "Cancelled" if context.custom_id == "modcase|no" else "Moderation case opened"
await context.send(msg)
await self.redis.delete(user_key)
await self.redis.delete(action_key)

20
jarvis/cogs/admin/__init__.py
View file

@ -1,12 +1,22 @@
"""JARVIS Admin Cogs."""
import logging
from dis_snek import Snake
from naff import Client
from jarvis.cogs.admin import ban, kick, lock, lockdown, mute, purge, roleping, warning
from jarvis.cogs.admin import (
ban,
kick,
lock,
lockdown,
modcase,
mute,
purge,
roleping,
warning,
)
def setup(bot: Snake) -> None:
def setup(bot: Client) -> None:
"""Add admin cogs to JARVIS"""
logger = logging.getLogger(__name__)
msg = "Loaded jarvis.cogs.admin.{}"
@ -17,7 +27,9 @@ def setup(bot: Snake) -> None:
lock.LockCog(bot)
logger.debug(msg.format("lock"))
lockdown.LockdownCog(bot)
logger.debug(msg.format("ban"))
logger.debug(msg.format("lockdown"))
modcase.CaseCog(bot)
logger.debug(msg.format("modcase"))
mute.MuteCog(bot)
logger.debug(msg.format("mute"))
purge.PurgeCog(bot)

23
jarvis/cogs/admin/ban.py
View file

@ -1,22 +1,21 @@
"""JARVIS BanCog."""
import logging
import re
from dis_snek import InteractionContext, Permissions, Snake
from dis_snek.client.utils.misc_utils import find, find_all
from dis_snek.ext.paginators import Paginator
from dis_snek.models.discord.embed import EmbedField
from dis_snek.models.discord.user import User
from dis_snek.models.snek.application_commands import (
from jarvis_core.db import q
from jarvis_core.db.models import Ban, Unban
from naff import InteractionContext, Permissions
from naff.client.utils.misc_utils import find, find_all
from naff.ext.paginators import Paginator
from naff.models.discord.embed import EmbedField
from naff.models.discord.user import User
from naff.models.naff.application_commands import (
OptionTypes,
SlashCommand,
SlashCommandChoice,
slash_command,
slash_option,
)
from dis_snek.models.snek.command import check
from jarvis_core.db import q
from jarvis_core.db.models import Ban, Unban
from naff.models.naff.command import check
from jarvis.utils import build_embed
from jarvis.utils.cogs import ModcaseCog
@ -26,10 +25,6 @@ from jarvis.utils.permissions import admin_or_permissions
class BanCog(ModcaseCog):
"""JARVIS BanCog."""
def __init__(self, bot: Snake):
super().__init__(bot)
self.logger = logging.getLogger(__name__)
async def discord_apply_ban(
self,
ctx: InteractionContext,

18
jarvis/cogs/admin/kick.py
View file

@ -1,16 +1,14 @@
"""JARVIS KickCog."""
import logging
from dis_snek import InteractionContext, Permissions, Snake
from dis_snek.models.discord.embed import EmbedField
from dis_snek.models.discord.user import User
from dis_snek.models.snek.application_commands import (
from jarvis_core.db.models import Kick
from naff import InteractionContext, Permissions
from naff.models.discord.embed import EmbedField
from naff.models.discord.user import User
from naff.models.naff.application_commands import (
OptionTypes,
slash_command,
slash_option,
)
from dis_snek.models.snek.command import check
from jarvis_core.db.models import Kick
from naff.models.naff.command import check
from jarvis.utils import build_embed
from jarvis.utils.cogs import ModcaseCog
@ -20,10 +18,6 @@ from jarvis.utils.permissions import admin_or_permissions
class KickCog(ModcaseCog):
"""JARVIS KickCog."""
def __init__(self, bot: Snake):
super().__init__(bot)
self.logger = logging.getLogger(__name__)
@slash_command(name="kick", description="Kick a user")
@slash_option(name="user", description="User to kick", opt_type=OptionTypes.USER, required=True)
@slash_option(

20
jarvis/cogs/admin/lock.py
View file

@ -2,26 +2,26 @@
import logging
from typing import Union
from dis_snek import InteractionContext, Scale, Snake
from dis_snek.client.utils.misc_utils import get
from dis_snek.models.discord.channel import GuildText, GuildVoice
from dis_snek.models.discord.enums import Permissions
from dis_snek.models.snek.application_commands import (
from jarvis_core.db import q
from jarvis_core.db.models import Lock, Permission
from naff import Client, Cog, InteractionContext
from naff.client.utils.misc_utils import get
from naff.models.discord.channel import GuildText, GuildVoice
from naff.models.discord.enums import Permissions
from naff.models.naff.application_commands import (
OptionTypes,
slash_command,
slash_option,
)
from dis_snek.models.snek.command import check
from jarvis_core.db import q
from jarvis_core.db.models import Lock, Permission
from naff.models.naff.command import check
from jarvis.utils.permissions import admin_or_permissions
class LockCog(Scale):
class LockCog(Cog):
"""JARVIS LockCog."""
def __init__(self, bot: Snake):
def __init__(self, bot: Client):
self.bot = bot
self.logger = logging.getLogger(__name__)

32
jarvis/cogs/admin/lockdown.py
View file

@ -1,25 +1,27 @@
"""JARVIS LockdownCog."""
import logging
from dis_snek import InteractionContext, Scale, Snake
from dis_snek.client.utils.misc_utils import find_all, get
from dis_snek.models.discord.channel import GuildCategory, GuildChannel
from dis_snek.models.discord.enums import Permissions
from dis_snek.models.discord.guild import Guild
from dis_snek.models.discord.user import Member
from dis_snek.models.snek.application_commands import (
from jarvis_core.db import q
from jarvis_core.db.models import Lock, Lockdown, Permission
from naff import Client, Cog, InteractionContext
from naff.client.utils.misc_utils import find_all, get
from naff.models.discord.channel import GuildCategory, GuildChannel
from naff.models.discord.enums import Permissions
from naff.models.discord.guild import Guild
from naff.models.discord.user import Member
from naff.models.naff.application_commands import (
OptionTypes,
SlashCommand,
slash_option,
)
from dis_snek.models.snek.command import check
from jarvis_core.db import q
from jarvis_core.db.models import Lock, Lockdown, Permission
from naff.models.naff.command import check
from jarvis.utils.permissions import admin_or_permissions
async def lock(bot: Snake, target: GuildChannel, admin: Member, reason: str, duration: int) -> None:
async def lock(
bot: Client, target: GuildChannel, admin: Member, reason: str, duration: int
) -> None:
"""
Lock an existing channel
@ -44,7 +46,7 @@ async def lock(bot: Snake, target: GuildChannel, admin: Member, reason: str, dur
).commit()
async def lock_all(bot: Snake, guild: Guild, admin: Member, reason: str, duration: int) -> None:
async def lock_all(bot: Client, guild: Guild, admin: Member, reason: str, duration: int) -> None:
"""
Lock all channels
@ -64,7 +66,7 @@ async def lock_all(bot: Snake, guild: Guild, admin: Member, reason: str, duratio
await lock(bot, channel, admin, reason, duration)
async def unlock_all(bot: Snake, guild: Guild, admin: Member) -> None:
async def unlock_all(bot: Client, guild: Guild, admin: Member) -> None:
"""
Unlock all locked channels
@ -92,10 +94,10 @@ async def unlock_all(bot: Snake, guild: Guild, admin: Member) -> None:
await lockdown.commit()
class LockdownCog(Scale):
class LockdownCog(Cog):
"""JARVIS LockdownCog."""
def __init__(self, bot: Snake):
def __init__(self, bot: Client):
self.bot = bot
self.logger = logging.getLogger(__name__)

332
jarvis/cogs/admin/modcase.py Normal file
View file

@ -0,0 +1,332 @@
"""JARVIS Moderation Case management."""
from typing import TYPE_CHECKING, List, Optional
from jarvis_core.db import q
from jarvis_core.db.models import Modlog, Note, actions
from naff import Cog, InteractionContext, Permissions
from naff.ext.paginators import Paginator
from naff.models.discord.embed import Embed, EmbedField
from naff.models.discord.user import Member
from naff.models.naff.application_commands import (
OptionTypes,
SlashCommand,
slash_option,
)
from naff.models.naff.command import check
from rich.console import Console
from rich.table import Table
from jarvis.utils import build_embed
from jarvis.utils.permissions import admin_or_permissions
if TYPE_CHECKING:
from naff.models.discord.guild import Guild
ACTIONS_LOOKUP = {
"ban": actions.Ban,
"kick": actions.Kick,
"mute": actions.Mute,
"unban": actions.Unban,
"warning": actions.Warning,
}
class CaseCog(Cog):
"""JARVIS CaseCog."""
async def get_summary_embed(self, mod_case: Modlog, guild: "Guild") -> Embed:
"""
Get Moderation case summary embed.
Args:
mod_case: Moderation case
guild: Originating guild
"""
action_table = Table()
action_table.add_column(header="Type", justify="left", style="orange4", no_wrap=True)
action_table.add_column(header="Admin", justify="left", style="cyan", no_wrap=True)
action_table.add_column(header="Reason", justify="left", style="white")
note_table = Table()
note_table.add_column(header="Admin", justify="left", style="cyan", no_wrap=True)
note_table.add_column(header="Content", justify="left", style="white")
console = Console()
action_output = ""
action_output_extra = ""
for idx, action in enumerate(mod_case.actions):
parent_action = await ACTIONS_LOOKUP[action.action_type].find_one(q(id=action.parent))
if not parent_action:
action.orphaned = True
action_table.add_row(action.action_type.title(), "[N/A]", "[N/A]")
else:
admin = await self.bot.fetch_user(parent_action.admin)
admin_text = "[N/A]"
if admin:
admin_text = f"{admin.username}#{admin.discriminator}"
action_table.add_row(action.action_type.title(), admin_text, parent_action.reason)
with console.capture() as cap:
console.print(action_table)
tmp_output = cap.get()
if len(tmp_output) >= 800:
action_output_extra = f"... and {len(mod_case.actions[idx:])} more actions"
break
action_output = tmp_output
note_output = ""
note_output_extra = ""
notes = sorted(mod_case.notes, key=lambda x: x.created_at)
for idx, note in enumerate(notes):
admin = await self.bot.fetch_user(note.admin)
admin_text = "[N/A]"
if admin:
admin_text = f"{admin.username}#{admin.discriminator}"
note_table.add_row(admin_text, note.content)
with console.capture() as cap:
console.print(note_table)
tmp_output = cap.get()
if len(tmp_output) >= 1000:
note_output_extra = f"... and {len(notes[idx:])} more notes"
break
note_output = tmp_output
status = "Open" if mod_case.open else "Closed"
user = await self.bot.fetch_user(mod_case.user)
username = "[N/A]"
user_text = "[N/A]"
if user:
username = f"{user.username}#{user.discriminator}"
user_text = user.mention
admin = await self.bot.fetch_user(mod_case.admin)
admin_text = "[N/A]"
if admin:
admin_text = admin.mention
action_output = f"```ansi\n{action_output}\n{action_output_extra}\n```"
note_output = f"```ansi\n{note_output}\n{note_output_extra}\n```"
fields = (
EmbedField(
name="Actions", value=action_output if mod_case.actions else "No Actions Found"
),
EmbedField(name="Notes", value=note_output if mod_case.notes else "No Notes Found"),
)
embed = build_embed(
title=f"Moderation Case [`{mod_case.nanoid}`]",
description=f"{status} case against {user_text} [**opened by {admin_text}**]",
fields=fields,
timestamp=mod_case.created_at,
)
icon_url = None
if user:
icon_url = user.avatar.url
embed.set_author(name=username, icon_url=icon_url)
embed.set_footer(text=str(mod_case.user))
await mod_case.commit()
return embed
async def get_action_embeds(self, mod_case: Modlog, guild: "Guild") -> List[Embed]:
"""
Get Moderation case action embeds.
Args:
mod_case: Moderation case
guild: Originating guild
"""
embeds = []
user = await self.bot.fetch_user(mod_case.user)
username = "[N/A]"
user_mention = "[N/A]"
avatar_url = None
if user:
username = f"{user.username}#{user.discriminator}"
avatar_url = user.avatar.url
user_mention = user.mention
for action in mod_case.actions:
if action.orphaned:
continue
parent_action = await ACTIONS_LOOKUP[action.action_type].find_one(q(id=action.parent))
if not parent_action:
action.orphaned = True
continue
admin = await self.bot.fetch_user(parent_action.admin)
admin_text = "[N/A]"
if admin:
admin_text = admin.mention
fields = (EmbedField(name=action.action_type.title(), value=parent_action.reason),)
embed = build_embed(
title="Moderation Case Action",
description=f"{admin_text} initiated an action against {user_mention}",
fields=fields,
timestamp=parent_action.created_at,
)
embed.set_author(name=username, icon_url=avatar_url)
embeds.append(embed)
await mod_case.commit()
return embeds
cases = SlashCommand(name="cases", description="Manage moderation cases")
@cases.subcommand(sub_cmd_name="list", sub_cmd_description="List moderation cases")
@slash_option(
name="user",
description="User to filter cases to",
opt_type=OptionTypes.USER,
required=False,
)
@slash_option(
name="closed",
description="Include closed cases",
opt_type=OptionTypes.BOOLEAN,
required=False,
)
@check(admin_or_permissions(Permissions.BAN_MEMBERS))
async def _cases_list(
self, ctx: InteractionContext, user: Optional[Member] = None, closed: bool = False
) -> None:
query = q(guild=ctx.guild.id)
if not closed:
query.update(q(open=True))
if user:
query.update(q(user=user.id))
cases = await Modlog.find(query).sort("created_at", -1).to_list(None)
if len(cases) == 0:
await ctx.send("No cases to view", ephemeral=True)
return
pages = [await self.get_summary_embed(c, ctx.guild) for c in cases]
paginator = Paginator.create_from_embeds(self.bot, *pages, timeout=300)
await paginator.send(ctx)
case = SlashCommand(name="case", description="Manage a moderation case")
show = case.group(name="show", description="Show information about a specific case")
@show.subcommand(sub_cmd_name="summary", sub_cmd_description="Summarize a specific case")
@slash_option(name="cid", description="Case ID", opt_type=OptionTypes.STRING, required=True)
@check(admin_or_permissions(Permissions.BAN_MEMBERS))
async def _case_show_summary(self, ctx: InteractionContext, cid: str) -> None:
case = await Modlog.find_one(q(guild=ctx.guild.id, nanoid=cid))
if not case:
await ctx.send(f"Could not find case with ID {cid}", ephemeral=True)
return
embed = await self.get_summary_embed(case, ctx.guild)
await ctx.send(embed=embed)
@show.subcommand(sub_cmd_name="actions", sub_cmd_description="Get case actions")
@slash_option(name="cid", description="Case ID", opt_type=OptionTypes.STRING, required=True)
@check(admin_or_permissions(Permissions.BAN_MEMBERS))
async def _case_show_actions(self, ctx: InteractionContext, cid: str) -> None:
case = await Modlog.find_one(q(guild=ctx.guild.id, nanoid=cid))
if not case:
await ctx.send(f"Could not find case with ID {cid}", ephemeral=True)
return
pages = await self.get_action_embeds(case, ctx.guild)
paginator = Paginator.create_from_embeds(self.bot, *pages, timeout=300)
await paginator.send(ctx)
@case.subcommand(sub_cmd_name="close", sub_cmd_description="Show a specific case")
@slash_option(name="cid", description="Case ID", opt_type=OptionTypes.STRING, required=True)
@check(admin_or_permissions(Permissions.BAN_MEMBERS))
async def _case_close(self, ctx: InteractionContext, cid: str) -> None:
case = await Modlog.find_one(q(guild=ctx.guild.id, nanoid=cid))
if not case:
await ctx.send(f"Could not find case with ID {cid}", ephemeral=True)
return
case.open = False
await case.commit()
embed = await self.get_summary_embed(case, ctx.guild)
await ctx.send(embed=embed)
@case.subcommand(sub_cmd_name="repoen", sub_cmd_description="Reopen a specific case")
@slash_option(name="cid", description="Case ID", opt_type=OptionTypes.STRING, required=True)
@check(admin_or_permissions(Permissions.BAN_MEMBERS))
async def _case_reopen(self, ctx: InteractionContext, cid: str) -> None:
case = await Modlog.find_one(q(guild=ctx.guild.id, nanoid=cid))
if not case:
await ctx.send(f"Could not find case with ID {cid}", ephemeral=True)
return
case.open = True
await case.commit()
embed = await self.get_summary_embed(case, ctx.guild)
await ctx.send(embed=embed)
@case.subcommand(sub_cmd_name="note", sub_cmd_description="Add a note to a specific case")
@slash_option(name="cid", description="Case ID", opt_type=OptionTypes.STRING, required=True)
@slash_option(
name="note", description="Note to add", opt_type=OptionTypes.STRING, required=True
)
@check(admin_or_permissions(Permissions.BAN_MEMBERS))
async def _case_note(self, ctx: InteractionContext, cid: str, note: str) -> None:
case = await Modlog.find_one(q(guild=ctx.guild.id, nanoid=cid))
if not case:
await ctx.send(f"Could not find case with ID {cid}", ephemeral=True)
return
if not case.open:
await ctx.send("Case is closed, please re-open to add a new comment", ephemeral=True)
return
if len(note) > 50:
await ctx.send("Note must be <= 50 characters", ephemeral=True)
return
note = Note(admin=ctx.author.id, content=note)
case.notes.append(note)
await case.commit()
embed = await self.get_summary_embed(case, ctx.guild)
await ctx.send(embed=embed)
@case.subcommand(sub_cmd_name="new", sub_cmd_description="Open a new case")
@slash_option(name="user", description="Target user", opt_type=OptionTypes.USER, required=True)
@slash_option(
name="note", description="Note to add", opt_type=OptionTypes.STRING, required=True
)
@check(admin_or_permissions(Permissions.BAN_MEMBERS))
async def _case_new(self, ctx: InteractionContext, user: Member, note: str) -> None:
case = await Modlog.find_one(q(guild=ctx.guild.id, user=user.id, open=True))
if case:
await ctx.send(f"Case already open with ID `{case.nanoid}`", ephemeral=True)
return
if not isinstance(user, Member):
await ctx.send("User must be in this guild", ephemeral=True)
return
if len(note) > 50:
await ctx.send("Note must be <= 50 characters", ephemeral=True)
return
note = Note(admin=ctx.author.id, content=note)
case = Modlog(
user=user.id, guild=ctx.guild.id, admin=ctx.author.id, notes=[note], actions=[]
)
await case.commit()
await case.reload()
embed = await self.get_summary_embed(case, ctx.guild)
await ctx.send(embed=embed)

26
jarvis/cogs/admin/mute.py
View file

@ -1,15 +1,16 @@
"""JARVIS MuteCog."""
import asyncio
import logging
from datetime import datetime, timedelta, timezone
from dateparser import parse
from dateparser_data.settings import default_parsers
from dis_snek import InteractionContext, Permissions, Snake
from dis_snek.models.discord.embed import EmbedField
from dis_snek.models.discord.modal import InputText, Modal, TextStyles
from dis_snek.models.discord.user import Member
from dis_snek.models.snek.application_commands import (
from jarvis_core.db.models import Mute
from naff import InteractionContext, Permissions
from naff.client.errors import Forbidden
from naff.models.discord.embed import EmbedField
from naff.models.discord.modal import InputText, Modal, TextStyles
from naff.models.discord.user import Member
from naff.models.naff.application_commands import (
CommandTypes,
OptionTypes,
SlashCommandChoice,
@ -17,8 +18,7 @@ from dis_snek.models.snek.application_commands import (
slash_command,
slash_option,
)
from dis_snek.models.snek.command import check
from jarvis_core.db.models import Mute
from naff.models.naff.command import check
from jarvis.utils import build_embed
from jarvis.utils.cogs import ModcaseCog
@ -28,10 +28,6 @@ from jarvis.utils.permissions import admin_or_permissions
class MuteCog(ModcaseCog):
"""JARVIS MuteCog."""
def __init__(self, bot: Snake):
super().__init__(bot)
self.logger = logging.getLogger(__name__)
async def _apply_timeout(
self, ctx: InteractionContext, user: Member, reason: str, until: datetime
) -> None:
@ -125,8 +121,11 @@ class MuteCog(ModcaseCog):
f"`{old_until}` is in the past, which isn't allowed", ephemeral=True
)
return
try:
embed = await self._apply_timeout(ctx, ctx.target, reason, until)
await response.send(embed=embed)
except Forbidden:
await response.send("Unable to mute this user", ephemeral=True)
@slash_command(name="mute", description="Mute a user")
@slash_option(name="user", description="User to mute", opt_type=OptionTypes.USER, required=True)
@ -179,8 +178,11 @@ class MuteCog(ModcaseCog):
return
until = datetime.now(tz=timezone.utc) + timedelta(minutes=duration)
try:
embed = await self._apply_timeout(ctx, user, reason, until)
await ctx.send(embed=embed)
except Forbidden:
await ctx.send("Unable to mute this user", ephemeral=True)
@slash_command(name="unmute", description="Unmute a user")
@slash_option(

16
jarvis/cogs/admin/purge.py
View file

@ -1,24 +1,24 @@
"""JARVIS PurgeCog."""
import logging
from dis_snek import InteractionContext, Permissions, Scale, Snake
from dis_snek.models.discord.channel import GuildText
from dis_snek.models.snek.application_commands import (
from jarvis_core.db import q
from jarvis_core.db.models import Autopurge, Purge
from naff import Client, Cog, InteractionContext, Permissions
from naff.models.discord.channel import GuildText
from naff.models.naff.application_commands import (
OptionTypes,
slash_command,
slash_option,
)
from dis_snek.models.snek.command import check
from jarvis_core.db import q
from jarvis_core.db.models import Autopurge, Purge
from naff.models.naff.command import check
from jarvis.utils.permissions import admin_or_permissions
class PurgeCog(Scale):
class PurgeCog(Cog):
"""JARVIS PurgeCog."""
def __init__(self, bot: Snake):
def __init__(self, bot: Client):
self.bot = bot
self.logger = logging.getLogger(__name__)

24
jarvis/cogs/admin/roleping.py
View file

@ -1,29 +1,29 @@
"""JARVIS RolepingCog."""
import logging
from dis_snek import InteractionContext, Permissions, Scale, Snake
from dis_snek.client.utils.misc_utils import find_all
from dis_snek.ext.paginators import Paginator
from dis_snek.models.discord.embed import EmbedField
from dis_snek.models.discord.role import Role
from dis_snek.models.discord.user import Member
from dis_snek.models.snek.application_commands import (
from jarvis_core.db import q
from jarvis_core.db.models import Roleping
from naff import Client, Cog, InteractionContext, Permissions
from naff.client.utils.misc_utils import find_all
from naff.ext.paginators import Paginator
from naff.models.discord.embed import EmbedField
from naff.models.discord.role import Role
from naff.models.discord.user import Member
from naff.models.naff.application_commands import (
OptionTypes,
SlashCommand,
slash_option,
)
from dis_snek.models.snek.command import check
from jarvis_core.db import q
from jarvis_core.db.models import Roleping
from naff.models.naff.command import check
from jarvis.utils import build_embed
from jarvis.utils.permissions import admin_or_permissions
class RolepingCog(Scale):
class RolepingCog(Cog):
"""JARVIS RolepingCog."""
def __init__(self, bot: Snake):
def __init__(self, bot: Client):
self.bot = bot
self.logger = logging.getLogger(__name__)

26
jarvis/cogs/admin/warning.py
View file

@ -1,19 +1,19 @@
"""JARVIS WarningCog."""
import logging
from datetime import datetime, timedelta, timezone
from dis_snek import InteractionContext, Permissions, Snake
from dis_snek.client.utils.misc_utils import get_all
from dis_snek.ext.paginators import Paginator
from dis_snek.models.discord.embed import EmbedField
from dis_snek.models.discord.user import Member
from dis_snek.models.snek.application_commands import (
from jarvis_core.db import q
from jarvis_core.db.models import Warning
from naff import InteractionContext, Permissions
from naff.client.utils.misc_utils import get_all
from naff.ext.paginators import Paginator
from naff.models.discord.embed import EmbedField
from naff.models.discord.user import Member
from naff.models.naff.application_commands import (
OptionTypes,
slash_command,
slash_option,
)
from dis_snek.models.snek.command import check
from jarvis_core.db import q
from jarvis_core.db.models import Warning
from naff.models.naff.command import check
from jarvis.utils import build_embed
from jarvis.utils.cogs import ModcaseCog
@ -24,10 +24,6 @@ from jarvis.utils.permissions import admin_or_permissions
class WarningCog(ModcaseCog):
"""JARVIS WarningCog."""
def __init__(self, bot: Snake):
super().__init__(bot)
self.logger = logging.getLogger(__name__)
@slash_command(name="warn", description="Warn a user")
@slash_option(name="user", description="User to warn", opt_type=OptionTypes.USER, required=True)
@slash_option(
@ -59,12 +55,14 @@ class WarningCog(ModcaseCog):
await ctx.send("User not in guild", ephemeral=True)
return
await ctx.defer()
expires_at = datetime.now(tz=timezone.utc) + timedelta(hours=duration)
await Warning(
user=user.id,
reason=reason,
admin=ctx.author.id,
guild=ctx.guild.id,
duration=duration,
expires_at=expires_at,
active=True,
).commit()
embed = warning_embed(user, reason)

20
jarvis/cogs/autoreact.py
View file

@ -3,26 +3,26 @@ import logging
import re
from typing import Optional, Tuple
from dis_snek import InteractionContext, Permissions, Scale, Snake
from dis_snek.client.utils.misc_utils import find
from dis_snek.models.discord.channel import GuildText
from dis_snek.models.snek.application_commands import (
from jarvis_core.db import q
from jarvis_core.db.models import Autoreact
from naff import Client, Cog, InteractionContext, Permissions
from naff.client.utils.misc_utils import find
from naff.models.discord.channel import GuildText
from naff.models.naff.application_commands import (
OptionTypes,
SlashCommand,
slash_option,
)
from dis_snek.models.snek.command import check
from jarvis_core.db import q
from jarvis_core.db.models import Autoreact
from naff.models.naff.command import check
from jarvis.data.unicode import emoji_list
from jarvis.utils.permissions import admin_or_permissions
class AutoReactCog(Scale):
class AutoReactCog(Cog):
"""JARVIS Autoreact Cog."""
def __init__(self, bot: Snake):
def __init__(self, bot: Client):
self.bot = bot
self.logger = logging.getLogger(__name__)
self.custom_emote = re.compile(r"^<:\w+:(\d+)>$")
@ -206,6 +206,6 @@ class AutoReactCog(Scale):
await ctx.send(message)
def setup(bot: Snake) -> None:
def setup(bot: Client) -> None:
"""Add AutoReactCog to JARVIS"""
AutoReactCog(bot)

96
jarvis/cogs/botutil.py
View file

@ -1,24 +1,33 @@
"""JARVIS bot utility commands."""
import logging
import platform
from io import BytesIO
import psutil
from aiofile import AIOFile, LineReader
from dis_snek import MessageContext, Scale, Snake
from dis_snek.models.discord.file import File
from molter import msg_command
from naff import Client, Cog, PrefixedContext, prefixed_command
from naff.models.discord.embed import EmbedField
from naff.models.discord.file import File
from rich.console import Console
from jarvis.utils import build_embed
from jarvis.utils.updates import update
class BotutilCog(Scale):
class BotutilCog(Cog):
"""JARVIS Bot Utility Cog."""
def __init__(self, bot: Snake):
def __init__(self, bot: Client):
self.bot = bot
self.logger = logging.getLogger(__name__)
self.add_cog_check(self.is_owner)
@msg_command(name="tail")
async def _tail(self, ctx: MessageContext, count: int = 10) -> None:
if ctx.author.id != self.bot.owner.id:
return
async def is_owner(self, ctx: PrefixedContext) -> bool:
"""Checks if author is bot owner."""
return ctx.author.id == self.bot.owner.id
@prefixed_command(name="tail")
async def _tail(self, ctx: PrefixedContext, count: int = 10) -> None:
lines = []
async with AIOFile("jarvis.log", "r") as af:
async for line in LineReader(af):
@ -35,11 +44,8 @@ class BotutilCog(Scale):
else:
await ctx.reply(content=f"```\n{log}\n```")
@msg_command(name="log")
async def _log(self, ctx: MessageContext) -> None:
if ctx.author.id != self.bot.owner.id:
return
@prefixed_command(name="log")
async def _log(self, ctx: PrefixedContext) -> None:
async with AIOFile("jarvis.log", "r") as af:
with BytesIO() as file_bytes:
raw = await af.read_bytes()
@ -48,7 +54,67 @@ class BotutilCog(Scale):
log = File(file_bytes, file_name="jarvis.log")
await ctx.reply(content="Here's the latest log", file=log)
@prefixed_command(name="crash")
async def _crash(self, ctx: PrefixedContext) -> None:
raise Exception("As you wish")
def setup(bot: Snake) -> None:
@prefixed_command(name="sysinfo")
async def _sysinfo(self, ctx: PrefixedContext) -> None:
st_ts = int(self.bot.start_time.timestamp())
ut_ts = int(psutil.boot_time())
fields = (
EmbedField(name="Operation System", value=platform.system() or "Unknown", inline=False),
EmbedField(name="Version", value=platform.release() or "N/A", inline=False),
EmbedField(name="System Start Time", value=f"<t:{ut_ts}:F> (<t:{ut_ts}:R>)"),
EmbedField(name="Python Version", value=platform.python_version()),
EmbedField(name="Bot Start Time", value=f"<t:{st_ts}:F> (<t:{st_ts}:R>)"),
)
embed = build_embed(title="System Info", description="", fields=fields)
embed.set_image(url=self.bot.user.avatar.url)
await ctx.send(embed=embed)
@prefixed_command(name="update")
async def _update(self, ctx: PrefixedContext) -> None:
status = await update(self.bot)
if status:
console = Console()
with console.capture() as capture:
console.print(status.table)
self.logger.debug(capture.get())
self.logger.debug(len(capture.get()))
added = "\n".join(status.added)
removed = "\n".join(status.removed)
changed = "\n".join(status.changed)
fields = [
EmbedField(name="Old Commit", value=status.old_hash),
EmbedField(name="New Commit", value=status.new_hash),
]
if added:
fields.append(EmbedField(name="New Modules", value=f"```\n{added}\n```"))
if removed:
fields.append(EmbedField(name="Removed Modules", value=f"```\n{removed}\n```"))
if changed:
fields.append(EmbedField(name="Changed Modules", value=f"```\n{changed}\n```"))
embed = build_embed(
"Update Status", description="Updates have been applied", fields=fields
)
embed.set_thumbnail(url="https://dev.zevaryx.com/git.png")
self.logger.info("Updates applied")
content = f"```ansi\n{capture.get()}\n```"
if len(content) < 3000:
await ctx.reply(content, embed=embed)
else:
await ctx.reply(f"Total Changes: {status.lines['total_lines']}", embed=embed)
else:
embed = build_embed(title="Update Status", description="No changes applied", fields=[])
embed.set_thumbnail(url="https://dev.zevaryx.com/git.png")
await ctx.reply(embed=embed)
def setup(bot: Client) -> None:
"""Add BotutilCog to JARVIS"""
BotutilCog(bot)

34
jarvis/cogs/ctc2.py
View file

@ -3,20 +3,20 @@ import logging
import re
import aiohttp
from dis_snek import InteractionContext, Scale, Snake
from dis_snek.ext.paginators import Paginator
from dis_snek.models.discord.components import ActionRow, Button, ButtonStyles
from dis_snek.models.discord.embed import EmbedField
from dis_snek.models.discord.user import Member, User
from dis_snek.models.snek.application_commands import (
from jarvis_core.db import q
from jarvis_core.db.models import Guess
from naff import Client, Cog, InteractionContext
from naff.ext.paginators import Paginator
from naff.models.discord.components import ActionRow, Button, ButtonStyles
from naff.models.discord.embed import EmbedField
from naff.models.discord.user import Member, User
from naff.models.naff.application_commands import (
OptionTypes,
SlashCommand,
slash_option,
)
from dis_snek.models.snek.command import cooldown
from dis_snek.models.snek.cooldowns import Buckets
from jarvis_core.db import q
from jarvis_core.db.models import Guess
from naff.models.naff.command import cooldown
from naff.models.naff.cooldowns import Buckets
from jarvis.utils import build_embed
@ -29,10 +29,10 @@ invites = re.compile(
)
class CTCCog(Scale):
class CTCCog(Cog):
"""JARVIS Complete the Code 2 Cog."""
def __init__(self, bot: Snake):
def __init__(self, bot: Client):
self.bot = bot
self.logger = logging.getLogger(__name__)
self._session = aiohttp.ClientSession()
@ -112,13 +112,11 @@ class CTCCog(Scale):
guesses = Guess.find().sort("correct", -1).sort("id", -1)
fields = []
async for guess in guesses:
user = await ctx.guild.get_member(guess["user"])
if not user:
user = await self.bot.fetch_user(guess["user"])
if not user:
user = "[redacted]"
if isinstance(user, (Member, User)):
user = user.name + "#" + user.discriminator
user = user.username + "#" + user.discriminator
name = "Correctly" if guess["correct"] else "Incorrectly"
name += " guessed by: " + user
fields.append(
@ -132,9 +130,9 @@ class CTCCog(Scale):
for i in range(0, len(fields), 5):
embed = build_embed(
title="completethecodetwo.cards guesses",
description=f"{len(fields)} guesses so far",
description=f"**{len(fields)} guesses so far**",
fields=fields[i : i + 5],
url="https://completethecodetwo.cards",
url="https://ctc2.zevaryx.com/gueses",
)
embed.set_thumbnail(url="https://dev.zevaryx.com/db_logo.png")
embed.set_footer(
@ -148,6 +146,6 @@ class CTCCog(Scale):
await paginator.send(ctx)
def setup(bot: Snake) -> None:
def setup(bot: Client) -> None:
"""Add CTCCog to JARVIS"""
CTCCog(bot)

20
jarvis/cogs/dbrand.py
View file

@ -3,37 +3,37 @@ import logging
import re
import aiohttp
from dis_snek import InteractionContext, Scale, Snake
from dis_snek.models.discord.embed import EmbedField
from dis_snek.models.snek.application_commands import (
from naff import Client, Cog, InteractionContext
from naff.models.discord.embed import EmbedField
from naff.models.naff.application_commands import (
OptionTypes,
SlashCommand,
slash_option,
)
from dis_snek.models.snek.command import cooldown
from dis_snek.models.snek.cooldowns import Buckets
from naff.models.naff.command import cooldown
from naff.models.naff.cooldowns import Buckets
from jarvis.config import get_config
from jarvis.config import JarvisConfig
from jarvis.data.dbrand import shipping_lookup
from jarvis.utils import build_embed
guild_ids = [862402786116763668] # [578757004059738142, 520021794380447745, 862402786116763668]
class DbrandCog(Scale):
class DbrandCog(Cog):
"""
dbrand functions for JARVIS
Mostly support functions. Credit @cpixl for the shipping API
"""
def __init__(self, bot: Snake):
def __init__(self, bot: Client):
self.bot = bot
self.logger = logging.getLogger(__name__)
self.base_url = "https://dbrand.com/"
self._session = aiohttp.ClientSession()
self._session.headers.update({"Content-Type": "application/json"})
self.api_url = get_config().urls["dbrand_shipping"]
self.api_url = JarvisConfig.from_yaml().urls["dbrand_shipping"]
self.cache = {}
def __del__(self):
@ -197,6 +197,6 @@ class DbrandCog(Scale):
await ctx.send(embed=embed)
def setup(bot: Snake) -> None:
def setup(bot: Client) -> None:
"""Add dbrandcog to JARVIS"""
DbrandCog(bot)

26
jarvis/cogs/dev.py
View file

@ -8,20 +8,20 @@ import uuid as uuidpy
import ulid as ulidpy
from bson import ObjectId
from dis_snek import InteractionContext, Scale, Snake
from dis_snek.models.discord.embed import EmbedField
from dis_snek.models.discord.message import Attachment
from dis_snek.models.snek.application_commands import (
from jarvis_core.filters import invites, url
from jarvis_core.util import convert_bytesize, hash
from jarvis_core.util.http import get_size
from naff import Client, Cog, InteractionContext
from naff.models.discord.embed import EmbedField
from naff.models.discord.message import Attachment
from naff.models.naff.application_commands import (
OptionTypes,
SlashCommandChoice,
slash_command,
slash_option,
)
from dis_snek.models.snek.command import cooldown
from dis_snek.models.snek.cooldowns import Buckets
from jarvis_core.filters import invites, url
from jarvis_core.util import convert_bytesize, hash
from jarvis_core.util.http import get_size
from naff.models.naff.command import cooldown
from naff.models.naff.cooldowns import Buckets
from jarvis.utils import build_embed
@ -45,10 +45,10 @@ UUID_GET = {3: uuidpy.uuid3, 5: uuidpy.uuid5}
MAX_FILESIZE = 5 * (1024**3) # 5GB
class DevCog(Scale):
class DevCog(Cog):
"""JARVIS Developer Cog."""
def __init__(self, bot: Snake):
def __init__(self, bot: Client):
self.bot = bot
self.logger = logging.getLogger(__name__)
@ -272,9 +272,9 @@ class DevCog(Scale):
output = subprocess.check_output( # noqa: S603, S607
["tokei", "-C", "--sort", "code"]
).decode("UTF-8")
await ctx.send(f"```\n{output}\n```")
await ctx.send(f"```haskell\n{output}\n```")
def setup(bot: Snake) -> None:
def setup(bot: Client) -> None:
"""Add DevCog to JARVIS"""
DevCog(bot)

28
jarvis/cogs/gl.py
View file

@ -4,20 +4,20 @@ import logging
from datetime import datetime
import gitlab
from dis_snek import InteractionContext, Scale, Snake
from dis_snek.ext.paginators import Paginator
from dis_snek.models.discord.embed import Embed, EmbedField
from dis_snek.models.discord.modal import InputText, Modal, TextStyles
from dis_snek.models.discord.user import Member
from dis_snek.models.snek.application_commands import (
from naff import Client, Cog, InteractionContext
from naff.ext.paginators import Paginator
from naff.models.discord.embed import Embed, EmbedField
from naff.models.discord.modal import InputText, Modal, TextStyles
from naff.models.discord.user import Member
from naff.models.naff.application_commands import (
OptionTypes,
SlashCommand,
SlashCommandChoice,
slash_command,
slash_option,
)
from dis_snek.models.snek.command import cooldown
from dis_snek.models.snek.cooldowns import Buckets
from naff.models.naff.command import cooldown
from naff.models.naff.cooldowns import Buckets
from jarvis.config import JarvisConfig
from jarvis.utils import build_embed
@ -25,10 +25,10 @@ from jarvis.utils import build_embed
guild_ids = [862402786116763668]
class GitlabCog(Scale):
class GitlabCog(Cog):
"""JARVIS GitLab Cog."""
def __init__(self, bot: Snake):
def __init__(self, bot: Client):
self.bot = bot
self.logger = logging.getLogger(__name__)
config = JarvisConfig.from_yaml()
@ -62,14 +62,16 @@ class GitlabCog(Scale):
labels = issue.labels
if labels:
labels = "\n".join(issue.labels)
if not labels:
else:
labels = "None"
fields = [
EmbedField(name="State", value=issue.state[0].upper() + issue.state[1:]),
EmbedField(name="State", value=issue.state.title()),
EmbedField(name="Assignee", value=assignee),
EmbedField(name="Labels", value=labels),
]
color = "#FC6D27"
if issue.labels:
color = self.project.labels.get(issue.labels[0]).color
fields.append(EmbedField(name="Created At", value=created_at))
if issue.state == "closed":
@ -463,7 +465,7 @@ class GitlabCog(Scale):
await resp.send(embed=embed)
def setup(bot: Snake) -> None:
def setup(bot: Client) -> None:
"""Add GitlabCog to JARVIS if Gitlab token exists."""
if JarvisConfig.from_yaml().gitlab_token:
GitlabCog(bot)

22
jarvis/cogs/image.py
View file

@ -6,28 +6,30 @@ from io import BytesIO
import aiohttp
import cv2
import numpy as np
from dis_snek import InteractionContext, Scale, Snake
from dis_snek.models.discord.embed import EmbedField
from dis_snek.models.discord.file import File
from dis_snek.models.discord.message import Attachment
from dis_snek.models.snek.application_commands import (
from jarvis_core.util import convert_bytesize, unconvert_bytesize
from naff import Client, Cog, InteractionContext
from naff.models.discord.embed import EmbedField
from naff.models.discord.file import File
from naff.models.discord.message import Attachment
from naff.models.naff.application_commands import (
OptionTypes,
slash_command,
slash_option,
)
from jarvis_core.util import build_embed, convert_bytesize, unconvert_bytesize
from jarvis.utils import build_embed
MIN_ACCURACY = 0.80
class ImageCog(Scale):
class ImageCog(Cog):
"""
Image processing functions for JARVIS
May be categorized under util later
"""
def __init__(self, bot: Snake):
def __init__(self, bot: Client):
self.bot = bot
self.logger = logging.getLogger(__name__)
self._session = aiohttp.ClientSession()
@ -83,7 +85,7 @@ class ImageCog(Scale):
if tgt_size > unconvert_bytesize(8, "MB"):
await ctx.send("Target too large to send. Please make target < 8MB", ephemeral=True)
return
elif tgt_size < 1024:
if tgt_size < 1024:
await ctx.send("Sizes < 1KB are extremely unreliable and are disabled", ephemeral=True)
return
@ -151,6 +153,6 @@ class ImageCog(Scale):
)
def setup(bot: Snake) -> None:
def setup(bot: Client) -> None:
"""Add ImageCog to JARVIS"""
ImageCog(bot)

425
jarvis/cogs/reddit.py Normal file
View file

@ -0,0 +1,425 @@
"""JARVIS Reddit cog."""
import asyncio
import logging
from typing import List, Optional
from asyncpraw import Reddit
from asyncpraw.models.reddit.submission import Submission
from asyncpraw.models.reddit.submission import Subreddit as Sub
from asyncprawcore.exceptions import Forbidden, NotFound, Redirect
from jarvis_core.db import q
from jarvis_core.db.models import Subreddit, SubredditFollow
from naff import Client, Cog, InteractionContext, Permissions
from naff.client.utils.misc_utils import get
from naff.models.discord.channel import ChannelTypes, GuildText
from naff.models.discord.components import ActionRow, Select, SelectOption
from naff.models.discord.embed import Embed, EmbedField
from naff.models.naff.application_commands import (
OptionTypes,
SlashCommand,
SlashCommandChoice,
slash_option,
)
from naff.models.naff.command import check
from jarvis import const
from jarvis.config import JarvisConfig
from jarvis.utils import build_embed
from jarvis.utils.permissions import admin_or_permissions
DEFAULT_USER_AGENT = f"python:JARVIS:{const.__version__} (by u/zevaryx)"
class RedditCog(Cog):
"""JARVIS Reddit Cog."""
def __init__(self, bot: Client):
self.bot = bot
self.logger = logging.getLogger(__name__)
config = JarvisConfig.from_yaml()
config.reddit["user_agent"] = config.reddit.get("user_agent", DEFAULT_USER_AGENT)
self.api = Reddit(**config.reddit)
async def post_embeds(self, sub: Sub, post: Submission) -> Optional[List[Embed]]:
"""
Build a post embeds.
Args:
post: Post to build embeds
"""
url = "https://reddit.com" + post.permalink
await post.author.load()
author_url = f"https://reddit.com/u/{post.author.name}"
author_icon = post.author.icon_img
images = []
title = f"{post.title}"
fields = []
content = ""
og_post = None
if not post.is_self:
og_post = post # noqa: F841
post = await self.api.submission(post.crosspost_parent_list[0]["id"])
await post.load()
fields.append(EmbedField(name="Crossposted From", value=post.subreddit_name_prefixed))
content = f"> **{post.title}**"
if "url" in vars(post):
if any(post.url.endswith(x) for x in ["jpeg", "jpg", "png", "gif"]):
images = [post.url]
if "media_metadata" in vars(post):
for k, v in post.media_metadata.items():
if v["status"] != "valid" or v["m"] not in ["image/jpg", "image/png", "image/gif"]:
continue
ext = v["m"].split("/")[-1]
i_url = f"https://i.redd.it/{k}.{ext}"
images.append(i_url)
if len(images) == 4:
break
if "selftext" in vars(post) and post.selftext:
content += "\n\n" + post.selftext
if len(content) > 900:
content = content[:900] + "..."
content += f"\n\n[View this post]({url})"
if not images and not content:
self.logger.debug(f"Post {post.id} had neither content nor images?")
return None
color = "#FF4500"
if "primary_color" in vars(sub):
color = sub.primary_color
base_embed = build_embed(
title=title,
description=content,
fields=fields,
timestamp=post.created_utc,
url=url,
color=color,
)
base_embed.set_author(
name="u/" + post.author.name, url=author_url, icon_url=author_icon
)
base_embed.set_footer(
text="Reddit", icon_url="https://www.redditinc.com/assets/images/site/reddit-logo.png"
)
embeds = [base_embed]
if len(images) > 0:
embeds[0].set_image(url=images[0])
for image in images[1:4]:
embed = Embed(url=url)
embed.set_image(url=image)
embeds.append(embed)
return embeds
reddit = SlashCommand(name="reddit", description="Manage Reddit follows")
@reddit.subcommand(sub_cmd_name="follow", sub_cmd_description="Follow a Subreddit")
@slash_option(
name="name",
description="Subreddit display name",
opt_type=OptionTypes.STRING,
required=True,
)
@slash_option(
name="channel",
description="Channel to post to",
opt_type=OptionTypes.CHANNEL,
channel_types=[ChannelTypes.GUILD_TEXT],
required=True,
)
@check(admin_or_permissions(Permissions.MANAGE_GUILD))
async def _reddit_follow(self, ctx: InteractionContext, name: str, channel: GuildText) -> None:
name = name.replace("r/", "")
if len(name) > 20 or len(name) < 3:
await ctx.send("Invalid Subreddit name", ephemeral=True)
return
if not isinstance(channel, GuildText):
await ctx.send("Channel must be a text channel", ephemeral=True)
return
try:
subreddit = await self.api.subreddit(name)
await subreddit.load()
except (NotFound, Forbidden, Redirect) as e:
self.logger.debug(f"Subreddit {name} raised {e.__class__.__name__} on add")
await ctx.send("Subreddit may be private, quarantined, or nonexistent.", ephemeral=True)
return
exists = await SubredditFollow.find_one(
q(display_name=subreddit.display_name, guild=ctx.guild.id)
)
if exists:
await ctx.send("Subreddit already being followed in this guild", ephemeral=True)
return
count = len([i async for i in SubredditFollow.find(q(guild=ctx.guild.id))])
if count >= 12:
await ctx.send("Cannot follow more than 12 Subreddits", ephemeral=True)
return
if subreddit.over18 and not channel.nsfw:
await ctx.send(
"Subreddit is nsfw, but channel is not. Mark the channel NSFW first.",
ephemeral=True,
)
return
sr = await Subreddit.find_one(q(display_name=subreddit.display_name))
if not sr:
sr = Subreddit(display_name=subreddit.display_name, over18=subreddit.over18)
await sr.commit()
srf = SubredditFollow(
display_name=subreddit.display_name,
channel=channel.id,
guild=ctx.guild.id,
admin=ctx.author.id,
)
await srf.commit()
await ctx.send(f"Now following `r/{name}` in {channel.mention}")
@reddit.subcommand(sub_cmd_name="unfollow", sub_cmd_description="Unfollow Subreddits")
@check(admin_or_permissions(Permissions.MANAGE_GUILD))
async def _subreddit_unfollow(self, ctx: InteractionContext) -> None:
subs = SubredditFollow.find(q(guild=ctx.guild.id))
subreddits = []
async for sub in subs:
subreddits.append(sub)
if not subreddits:
await ctx.send("You need to follow a Subreddit first", ephemeral=True)
return
options = []
names = []
for idx, subreddit in enumerate(subreddits):
sub = await Subreddit.find_one(q(display_name=subreddit.display_name))
names.append(sub.display_name)
option = SelectOption(label=sub.display_name, value=str(idx))
options.append(option)
select = Select(
options=options, custom_id="to_delete", min_values=1, max_values=len(subreddits)
)
components = [ActionRow(select)]
block = "\n".join(x for x in names)
message = await ctx.send(
content=f"You are following the following subreddits:\n```\n{block}\n```\n\n"
"Please choose subreddits to unfollow",
components=components,
)
try:
context = await self.bot.wait_for_component(
check=lambda x: ctx.author.id == x.context.author.id,
messages=message,
timeout=60 * 5,
)
for to_delete in context.context.values:
follow = get(subreddits, guild=ctx.guild.id, display_name=names[int(to_delete)])
try:
await follow.delete()
except Exception:
self.logger.debug("Ignoring deletion error")
for row in components:
for component in row.components:
component.disabled = True
block = "\n".join(names[int(x)] for x in context.context.values)
await context.context.edit_origin(
content=f"Unfollowed the following:\n```\n{block}\n```", components=components
)
except asyncio.TimeoutError:
for row in components:
for component in row.components:
component.disabled = True
await message.edit(components=components)
@reddit.subcommand(sub_cmd_name="hot", sub_cmd_description="Get the hot post of a subreddit")
@slash_option(
name="name", description="Subreddit name", opt_type=OptionTypes.STRING, required=True
)
async def _subreddit_hot(self, ctx: InteractionContext, name: str) -> None:
await ctx.defer()
name = name.replace("r/", "")
if len(name) > 20 or len(name) < 3:
await ctx.send("Invalid Subreddit name", ephemeral=True)
return
try:
subreddit = await self.api.subreddit(name)
await subreddit.load()
except (NotFound, Forbidden, Redirect) as e:
self.logger.debug(f"Subreddit {name} raised {e.__class__.__name__} in hot")
await ctx.send("Subreddit may be private, quarantined, or nonexistent.", ephemeral=True)
return
try:
post = [x async for x in subreddit.hot(limit=1)][0]
except Exception as e:
self.logger.error(f"Failed to get post from {name}", exc_info=e)
await ctx.send("Well, this is awkward. Something went wrong", ephemeral=True)
return
embeds = await self.post_embeds(subreddit, post)
if post.over_18 and not ctx.channel.nsfw:
try:
await ctx.author.send(embeds=embeds)
await ctx.send("Hey! Due to content, I had to DM the result to you")
except Exception:
await ctx.send("Hey! Due to content, I cannot share the result")
else:
await ctx.send(embeds=embeds)
@reddit.subcommand(sub_cmd_name="top", sub_cmd_description="Get the top post of a subreddit")
@slash_option(
name="name", description="Subreddit name", opt_type=OptionTypes.STRING, required=True
)
@slash_option(
name="time",
description="Top time",
opt_type=OptionTypes.STRING,
required=False,
choices=[
SlashCommandChoice(name="All", value="all"),
SlashCommandChoice(name="Day", value="day"),
SlashCommandChoice(name="Hour", value="hour"),
SlashCommandChoice(name="Month", value="month"),
SlashCommandChoice(name="Week", value="week"),
SlashCommandChoice(name="Year", value="year"),
],
)
async def _subreddit_top(self, ctx: InteractionContext, name: str, time: str = "all") -> None:
await ctx.defer()
name = name.replace("r/", "")
if len(name) > 20 or len(name) < 3:
await ctx.send("Invalid Subreddit name", ephemeral=True)
return
try:
subreddit = await self.api.subreddit(name)
await subreddit.load()
except (NotFound, Forbidden, Redirect) as e:
self.logger.debug(f"Subreddit {name} raised {e.__class__.__name__} in top")
await ctx.send("Subreddit may be private, quarantined, or nonexistent.", ephemeral=True)
return
try:
post = [x async for x in subreddit.top(time_filter=time, limit=1)][0]
except Exception as e:
self.logger.error(f"Failed to get post from {name}", exc_info=e)
await ctx.send("Well, this is awkward. Something went wrong", ephemeral=True)
return
embeds = await self.post_embeds(subreddit, post)
if post.over_18 and not ctx.channel.nsfw:
try:
await ctx.author.send(embeds=embeds)
await ctx.send("Hey! Due to content, I had to DM the result to you")
except Exception:
await ctx.send("Hey! Due to content, I cannot share the result")
else:
await ctx.send(embeds=embeds)
@reddit.subcommand(
sub_cmd_name="random", sub_cmd_description="Get a random post of a subreddit"
)
@slash_option(
name="name", description="Subreddit name", opt_type=OptionTypes.STRING, required=True
)
async def _subreddit_random(self, ctx: InteractionContext, name: str) -> None:
await ctx.defer()
name = name.replace("r/", "")
if len(name) > 20 or len(name) < 3:
await ctx.send("Invalid Subreddit name", ephemeral=True)
return
try:
subreddit = await self.api.subreddit(name)
await subreddit.load()
except (NotFound, Forbidden, Redirect) as e:
self.logger.debug(f"Subreddit {name} raised {e.__class__.__name__} in random")
await ctx.send("Subreddit may be private, quarantined, or nonexistent.", ephemeral=True)
return
try:
post = await subreddit.random()
except Exception as e:
self.logger.error(f"Failed to get post from {name}", exc_info=e)
await ctx.send("Well, this is awkward. Something went wrong", ephemeral=True)
return
embeds = await self.post_embeds(subreddit, post)
if post.over_18 and not ctx.channel.nsfw:
try:
await ctx.author.send(embeds=embeds)
await ctx.send("Hey! Due to content, I had to DM the result to you")
except Exception:
await ctx.send("Hey! Due to content, I cannot share the result")
else:
await ctx.send(embeds=embeds)
@reddit.subcommand(
sub_cmd_name="rising", sub_cmd_description="Get a rising post of a subreddit"
)
@slash_option(
name="name", description="Subreddit name", opt_type=OptionTypes.STRING, required=True
)
async def _subreddit_rising(self, ctx: InteractionContext, name: str) -> None:
await ctx.defer()
name = name.replace("r/", "")
if len(name) > 20 or len(name) < 3:
await ctx.send("Invalid Subreddit name", ephemeral=True)
return
try:
subreddit = await self.api.subreddit(name)
await subreddit.load()
except (NotFound, Forbidden, Redirect) as e:
self.logger.debug(f"Subreddit {name} raised {e.__class__.__name__} in rising")
await ctx.send("Subreddit may be private, quarantined, or nonexistent.", ephemeral=True)
return
try:
post = [x async for x in subreddit.rising(limit=1)][0]
except Exception as e:
self.logger.error(f"Failed to get post from {name}", exc_info=e)
await ctx.send("Well, this is awkward. Something went wrong", ephemeral=True)
return
embeds = await self.post_embeds(subreddit, post)
if post.over_18 and not ctx.channel.nsfw:
try:
await ctx.author.send(embeds=embeds)
await ctx.send("Hey! Due to content, I had to DM the result to you")
except Exception:
await ctx.send("Hey! Due to content, I cannot share the result")
else:
await ctx.send(embeds=embeds)
@reddit.subcommand(sub_cmd_name="post", sub_cmd_description="Get a specific submission")
@slash_option(
name="sid", description="Submission ID", opt_type=OptionTypes.STRING, required=True
)
async def _reddit_post(self, ctx: InteractionContext, sid: str) -> None:
await ctx.defer()
try:
post = await self.api.submission(sid)
await post.load()
except (NotFound, Forbidden, Redirect) as e:
self.logger.debug(f"Submission {sid} raised {e.__class__.__name__} in post")
await ctx.send("Subreddit may be private, quarantined, or nonexistent.", ephemeral=True)
return
embeds = await self.post_embeds(post.subreddit, post)
if post.over_18 and not ctx.channel.nsfw:
try:
await ctx.author.send(embeds=embeds)
await ctx.send("Hey! Due to content, I had to DM the result to you")
except Exception:
await ctx.send("Hey! Due to content, I cannot share the result")
else:
await ctx.send(embeds=embeds)
def setup(bot: Client) -> None:
"""Add RedditCog to JARVIS"""
if JarvisConfig.from_yaml().reddit:
RedditCog(bot)

26
jarvis/cogs/remindme.py
View file

@ -8,20 +8,20 @@ from typing import List
from bson import ObjectId
from dateparser import parse
from dateparser_data.settings import default_parsers
from dis_snek import InteractionContext, Scale, Snake
from dis_snek.client.utils.misc_utils import get
from dis_snek.models.discord.channel import GuildChannel
from dis_snek.models.discord.components import ActionRow, Select, SelectOption
from dis_snek.models.discord.embed import Embed, EmbedField
from dis_snek.models.discord.modal import InputText, Modal, TextStyles
from dis_snek.models.snek.application_commands import (
from jarvis_core.db import q
from jarvis_core.db.models import Reminder
from naff import Client, Cog, InteractionContext
from naff.client.utils.misc_utils import get
from naff.models.discord.channel import GuildChannel
from naff.models.discord.components import ActionRow, Select, SelectOption
from naff.models.discord.embed import Embed, EmbedField
from naff.models.discord.modal import InputText, Modal, TextStyles
from naff.models.naff.application_commands import (
OptionTypes,
SlashCommand,
slash_command,
slash_option,
)
from jarvis_core.db import q
from jarvis_core.db.models import Reminder
from jarvis.utils import build_embed
@ -33,10 +33,10 @@ invites = re.compile(
)
class RemindmeCog(Scale):
class RemindmeCog(Cog):
"""JARVIS Remind Me Cog."""
def __init__(self, bot: Snake):
def __init__(self, bot: Client):
self.bot = bot
self.logger = logging.getLogger(__name__)
@ -329,13 +329,13 @@ class RemindmeCog(Scale):
embed.set_thumbnail(url=ctx.author.display_avatar)
await ctx.send(embed=embed, ephemeral=reminder.private)
if reminder.remind_at <= datetime.now(tz=timezone.utc):
if reminder.remind_at <= datetime.now(tz=timezone.utc) and not reminder.active:
try:
await reminder.delete()
except Exception:
self.logger.debug("Ignoring deletion error")
def setup(bot: Snake) -> None:
def setup(bot: Client) -> None:
"""Add RemindmeCog to JARVIS"""
RemindmeCog(bot)

37
jarvis/cogs/rolegiver.py
View file

@ -2,29 +2,29 @@
import asyncio
import logging
from dis_snek import InteractionContext, Permissions, Scale, Snake
from dis_snek.client.utils.misc_utils import get
from dis_snek.models.discord.components import ActionRow, Select, SelectOption
from dis_snek.models.discord.embed import EmbedField
from dis_snek.models.discord.role import Role
from dis_snek.models.snek.application_commands import (
from jarvis_core.db import q
from jarvis_core.db.models import Rolegiver
from naff import Client, Cog, InteractionContext, Permissions
from naff.client.utils.misc_utils import get
from naff.models.discord.components import ActionRow, Select, SelectOption
from naff.models.discord.embed import EmbedField
from naff.models.discord.role import Role
from naff.models.naff.application_commands import (
OptionTypes,
SlashCommand,
slash_option,
)
from dis_snek.models.snek.command import check, cooldown
from dis_snek.models.snek.cooldowns import Buckets
from jarvis_core.db import q
from jarvis_core.db.models import Rolegiver
from naff.models.naff.command import check, cooldown
from naff.models.naff.cooldowns import Buckets
from jarvis.utils import build_embed
from jarvis.utils.permissions import admin_or_permissions
class RolegiverCog(Scale):
class RolegiverCog(Cog):
"""JARVIS Role Giver Cog."""
def __init__(self, bot: Snake):
def __init__(self, bot: Client):
self.bot = bot
self.logger = logging.getLogger(__name__)
@ -41,14 +41,23 @@ class RolegiverCog(Scale):
await ctx.send("Cannot add `@everyone` to rolegiver", ephemeral=True)
return
if role.bot_managed or not role.is_assignable:
await ctx.send(
"Cannot assign this role, try lowering it below my role or using a different role",
ephemeral=True,
)
return
setting = await Rolegiver.find_one(q(guild=ctx.guild.id))
if setting and role.id in setting.roles:
if setting and setting.roles and role.id in setting.roles:
await ctx.send("Role already in rolegiver", ephemeral=True)
return
if not setting:
setting = Rolegiver(guild=ctx.guild.id, roles=[])
setting.roles = setting.roles or []
if len(setting.roles) >= 20:
await ctx.send("You can only have 20 roles in the rolegiver", ephemeral=True)
return
@ -378,6 +387,6 @@ class RolegiverCog(Scale):
await ctx.send("Rolegiver cleanup finished")
def setup(bot: Snake) -> None:
def setup(bot: Client) -> None:
"""Add RolegiverCog to JARVIS"""
RolegiverCog(bot)

47
jarvis/cogs/settings.py
View file

@ -3,29 +3,29 @@ import asyncio
import logging
from typing import Any
from dis_snek import InteractionContext, Scale, Snake
from dis_snek.models.discord.channel import GuildText
from dis_snek.models.discord.components import ActionRow, Button, ButtonStyles
from dis_snek.models.discord.embed import EmbedField
from dis_snek.models.discord.enums import Permissions
from dis_snek.models.discord.role import Role
from dis_snek.models.snek.application_commands import (
from jarvis_core.db import q
from jarvis_core.db.models import Setting
from naff import Client, Cog, InteractionContext
from naff.models.discord.channel import GuildText
from naff.models.discord.components import ActionRow, Button, ButtonStyles
from naff.models.discord.embed import EmbedField
from naff.models.discord.enums import Permissions
from naff.models.discord.role import Role
from naff.models.naff.application_commands import (
OptionTypes,
SlashCommand,
slash_option,
)
from dis_snek.models.snek.command import check
from jarvis_core.db import q
from jarvis_core.db.models import Setting
from naff.models.naff.command import check
from jarvis.utils import build_embed
from jarvis.utils.permissions import admin_or_permissions
class SettingsCog(Scale):
class SettingsCog(Cog):
"""JARVIS Settings Management Cog."""
def __init__(self, bot: Snake):
def __init__(self, bot: Client):
self.bot = bot
self.logger = logging.getLogger(__name__)
@ -105,6 +105,12 @@ class SettingsCog(Scale):
if role.id == ctx.guild.id:
await ctx.send("Cannot set verified to `@everyone`", ephemeral=True)
return
if role.bot_managed or not role.is_assignable:
await ctx.send(
"Cannot assign this role, try lowering it below my role or using a different role",
ephemeral=True,
)
return
await ctx.defer()
await self.update_settings("verified", role.id, ctx.guild.id)
await ctx.send(f"Settings applied. New verified role is `{role.name}`")
@ -118,6 +124,12 @@ class SettingsCog(Scale):
if role.id == ctx.guild.id:
await ctx.send("Cannot set unverified to `@everyone`", ephemeral=True)
return
if role.bot_managed or not role.is_assignable:
await ctx.send(
"Cannot assign this role, try lowering it below my role or using a different role",
ephemeral=True,
)
return
await ctx.defer()
await self.update_settings("unverified", role.id, ctx.guild.id)
await ctx.send(f"Settings applied. New unverified role is `{role.name}`")
@ -169,14 +181,11 @@ class SettingsCog(Scale):
await ctx.send("Setting `massmention` unset")
@unset.subcommand(sub_cmd_name="verified", sub_cmd_description="Unset verified role")
@slash_option(
name="role", description="Verified role", opt_type=OptionTypes.ROLE, required=True
)
@check(admin_or_permissions(Permissions.MANAGE_GUILD))
async def _unset_verified(self, ctx: InteractionContext) -> None:
await ctx.defer()
await self.delete_settings("verified", ctx.guild.id)
await ctx.send("Setting `massmention` unset")
await ctx.send("Setting `verified` unset")
@unset.subcommand(sub_cmd_name="unverified", sub_cmd_description="Unset unverified role")
@check(admin_or_permissions(Permissions.MANAGE_GUILD))
@ -210,7 +219,11 @@ class SettingsCog(Scale):
async for setting in settings:
value = setting.value
if setting.setting in ["unverified", "verified", "mute"]:
try:
value = await ctx.guild.fetch_role(value)
except KeyError:
await setting.delete()
continue
if value:
value = value.mention
else:
@ -269,6 +282,6 @@ class SettingsCog(Scale):
await message.edit(content="Guild settings not cleared", components=components)
def setup(bot: Snake) -> None:
def setup(bot: Client) -> None:
"""Add SettingsCog to JARVIS"""
SettingsCog(bot)

24
jarvis/cogs/starboard.py
View file

@ -1,20 +1,20 @@
"""JARVIS Starboard Cog."""
import logging
from dis_snek import InteractionContext, Permissions, Scale, Snake
from dis_snek.models.discord.channel import GuildText
from dis_snek.models.discord.components import ActionRow, Select, SelectOption
from dis_snek.models.discord.message import Message
from dis_snek.models.snek.application_commands import (
from jarvis_core.db import q
from jarvis_core.db.models import Star, Starboard
from naff import Client, Cog, InteractionContext, Permissions
from naff.models.discord.channel import GuildText
from naff.models.discord.components import ActionRow, Select, SelectOption
from naff.models.discord.message import Message
from naff.models.naff.application_commands import (
CommandTypes,
OptionTypes,
SlashCommand,
context_menu,
slash_option,
)
from dis_snek.models.snek.command import check
from jarvis_core.db import q
from jarvis_core.db.models import Star, Starboard
from naff.models.naff.command import check
from jarvis.utils import build_embed
from jarvis.utils.permissions import admin_or_permissions
@ -28,10 +28,10 @@ supported_images = [
]
class StarboardCog(Scale):
class StarboardCog(Cog):
"""JARVIS Starboard Cog."""
def __init__(self, bot: Snake):
def __init__(self, bot: Client):
self.bot = bot
self.logger = logging.getLogger(__name__)
@ -135,7 +135,7 @@ class StarboardCog(Scale):
if c and isinstance(c, GuildText):
channel_list.append(c)
else:
self.logger.warn(
self.logger.warning(
f"Starboard {starboard.channel} no longer valid in {ctx.guild.name}"
)
to_delete.append(starboard)
@ -318,6 +318,6 @@ class StarboardCog(Scale):
await ctx.send(f"Star {id} deleted from {starboard.mention}")
def setup(bot: Snake) -> None:
def setup(bot: Client) -> None:
"""Add StarboardCog to JARVIS"""
StarboardCog(bot)

126
jarvis/cogs/temprole.py Normal file
View file

@ -0,0 +1,126 @@
"""JARVIS temporary role handler."""
import logging
from datetime import datetime, timezone
from dateparser import parse
from dateparser_data.settings import default_parsers
from jarvis_core.db.models import Temprole
from naff import Client, Cog, InteractionContext, Permissions
from naff.models.discord.embed import EmbedField
from naff.models.discord.role import Role
from naff.models.discord.user import Member
from naff.models.naff.application_commands import (
OptionTypes,
slash_command,
slash_option,
)
from naff.models.naff.command import check
from jarvis.utils import build_embed
from jarvis.utils.permissions import admin_or_permissions
class TemproleCog(Cog):
"""JARVIS Temporary Role Cog."""
def __init__(self, bot: Client):
self.bot = bot
self.logger = logging.getLogger(__name__)
@slash_command(name="temprole", description="Give a user a temporary role")
@slash_option(
name="user", description="User to grant role", opt_type=OptionTypes.USER, required=True
)
@slash_option(
name="role", description="Role to grant", opt_type=OptionTypes.ROLE, required=True
)
@slash_option(
name="duration",
description="Duration of temp role (i.e. 2 hours)",
opt_type=OptionTypes.STRING,
required=True,
)
@slash_option(
name="reason",
description="Reason for temporary role",
opt_type=OptionTypes.STRING,
required=False,
)
@check(admin_or_permissions(Permissions.MANAGE_ROLES))
async def _temprole(
self, ctx: InteractionContext, user: Member, role: Role, duration: str, reason: str = None
) -> None:
await ctx.defer()
if not isinstance(user, Member):
await ctx.send("User not in guild", ephemeral=True)
return
if role.id == ctx.guild.id:
await ctx.send("Cannot add `@everyone` to users", ephemeral=True)
return
if role.bot_managed or not role.is_assignable:
await ctx.send(
"Cannot assign this role, try lowering it below my role or using a different role",
ephemeral=True,
)
return
base_settings = {
"PREFER_DATES_FROM": "future",
"TIMEZONE": "UTC",
"RETURN_AS_TIMEZONE_AWARE": True,
}
rt_settings = base_settings.copy()
rt_settings["PARSERS"] = [
x for x in default_parsers if x not in ["absolute-time", "timestamp"]
]
rt_duration = parse(duration, settings=rt_settings)
at_settings = base_settings.copy()
at_settings["PARSERS"] = [x for x in default_parsers if x != "relative-time"]
at_duration = parse(duration, settings=at_settings)
if rt_duration:
duration = rt_duration
elif at_duration:
duration = at_duration
else:
self.logger.debug(f"Failed to parse duration: {duration}")
await ctx.send(f"`{duration}` is not a parsable date, please try again", ephemeral=True)
return
if duration < datetime.now(tz=timezone.utc):
await ctx.send(
f"`{duration}` is in the past. Past durations aren't allowed", ephemeral=True
)
return
await user.add_role(role, reason=reason)
await Temprole(
guild=ctx.guild.id, user=user.id, role=role.id, admin=ctx.author.id, expires_at=duration
).commit()
ts = int(duration.timestamp())
fields = (
EmbedField(name="Role", value=role.mention),
EmbedField(name="Valid Until", value=f"<t:{ts}:F> (<t:{ts}:R>)"),
)
embed = build_embed(
title="Role granted",
description=f"Role temporarily granted to {user.mention}",
fields=fields,
)
embed.set_author(
name=f"{user.username}#{user.discriminator}", icon_url=user.display_avatar.url
)
await ctx.send(embed=embed)
def setup(bot: Client) -> None:
"""Add TemproleCog to JARVIS"""
TemproleCog(bot)

35
jarvis/cogs/twitter.py
View file

@ -3,27 +3,27 @@ import asyncio
import logging
import tweepy
from dis_snek import InteractionContext, Permissions, Scale, Snake
from dis_snek.client.utils.misc_utils import get
from dis_snek.models.discord.channel import GuildText
from dis_snek.models.discord.components import ActionRow, Select, SelectOption
from dis_snek.models.snek.application_commands import (
from jarvis_core.db import q
from jarvis_core.db.models import TwitterAccount, TwitterFollow
from naff import Client, Cog, InteractionContext, Permissions
from naff.client.utils.misc_utils import get
from naff.models.discord.channel import GuildText
from naff.models.discord.components import ActionRow, Select, SelectOption
from naff.models.naff.application_commands import (
OptionTypes,
SlashCommand,
slash_option,
)
from dis_snek.models.snek.command import check
from jarvis_core.db import q
from jarvis_core.db.models import TwitterAccount, TwitterFollow
from naff.models.naff.command import check
from jarvis.config import JarvisConfig
from jarvis.utils.permissions import admin_or_permissions
class TwitterCog(Scale):
class TwitterCog(Cog):
"""JARVIS Twitter Cog."""
def __init__(self, bot: Snake):
def __init__(self, bot: Client):
self.bot = bot
self.logger = logging.getLogger(__name__)
config = JarvisConfig.from_yaml()
@ -63,7 +63,7 @@ class TwitterCog(Scale):
self, ctx: InteractionContext, handle: str, channel: GuildText, retweets: bool = True
) -> None:
handle = handle.lower()
if len(handle) > 15:
if len(handle) > 15 or len(handle) < 4:
await ctx.send("Invalid Twitter handle", ephemeral=True)
return
@ -80,16 +80,16 @@ class TwitterCog(Scale):
)
return
count = len([i async for i in TwitterFollow.find(q(guild=ctx.guild.id))])
if count >= 12:
await ctx.send("Cannot follow more than 12 Twitter accounts", ephemeral=True)
return
exists = await TwitterFollow.find_one(q(twitter_id=account.id, guild=ctx.guild.id))
if exists:
await ctx.send("Twitter account already being followed in this guild", ephemeral=True)
return
count = len([i async for i in TwitterFollow.find(q(guild=ctx.guild.id))])
if count >= 12:
await ctx.send("Cannot follow more than 12 Twitter accounts", ephemeral=True)
return
ta = await TwitterAccount.find_one(q(twitter_id=account.id))
if not ta:
ta = TwitterAccount(
@ -243,6 +243,7 @@ class TwitterCog(Scale):
await message.edit(components=components)
def setup(bot: Snake) -> None:
def setup(bot: Client) -> None:
"""Add TwitterCog to JARVIS"""
if JarvisConfig.from_yaml().twitter:
TwitterCog(bot)

46
jarvis/cogs/util.py
View file

@ -1,6 +1,5 @@
"""JARVIS Utility Cog."""
import logging
import platform
import re
import secrets
import string
@ -9,14 +8,14 @@ from io import BytesIO
import numpy as np
from dateparser import parse
from dis_snek import InteractionContext, Scale, Snake, const
from dis_snek.models.discord.channel import GuildCategory, GuildText, GuildVoice
from dis_snek.models.discord.embed import EmbedField
from dis_snek.models.discord.file import File
from dis_snek.models.discord.guild import Guild
from dis_snek.models.discord.role import Role
from dis_snek.models.discord.user import Member, User
from dis_snek.models.snek.application_commands import (
from naff import Client, Cog, InteractionContext, const
from naff.models.discord.channel import GuildCategory, GuildText, GuildVoice
from naff.models.discord.embed import EmbedField
from naff.models.discord.file import File
from naff.models.discord.guild import Guild
from naff.models.discord.role import Role
from naff.models.discord.user import Member, User
from naff.models.naff.application_commands import (
CommandTypes,
OptionTypes,
SlashCommandChoice,
@ -24,12 +23,12 @@ from dis_snek.models.snek.application_commands import (
slash_command,
slash_option,
)
from dis_snek.models.snek.command import cooldown
from dis_snek.models.snek.cooldowns import Buckets
from naff.models.naff.command import cooldown
from naff.models.naff.cooldowns import Buckets
from PIL import Image
from tzlocal import get_localzone
import jarvis
from jarvis import const as jconst
from jarvis.data import pigpen
from jarvis.data.robotcamo import emotes, hk, names
from jarvis.utils import build_embed, get_repo_hash
@ -37,14 +36,14 @@ from jarvis.utils import build_embed, get_repo_hash
JARVIS_LOGO = Image.open("jarvis_small.png").convert("RGBA")
class UtilCog(Scale):
class UtilCog(Cog):
"""
Utility functions for JARVIS
Mostly system utility functions, but may change over time
"""
def __init__(self, bot: Snake):
def __init__(self, bot: Client):
self.bot = bot
self.logger = logging.getLogger(__name__)
@ -55,10 +54,12 @@ class UtilCog(Scale):
desc = f"All systems online\nConnected to **{len(self.bot.guilds)}** guilds"
color = "#3498db"
fields = []
uptime = int(self.bot.start_time.timestamp())
fields.append(EmbedField(name="dis-snek", value=const.__version__))
fields.append(EmbedField(name="Version", value=jarvis.__version__, inline=False))
fields.append(EmbedField(name="Git Hash", value=get_repo_hash()[:7], inline=False))
fields.append(EmbedField(name="Version", value=jconst.__version__, inline=True))
fields.append(EmbedField(name="naff", value=const.__version__, inline=True))
fields.append(EmbedField(name="Git Hash", value=get_repo_hash()[:7], inline=True))
fields.append(EmbedField(name="Online Since", value=f"<t:{uptime}:F>", inline=False))
num_domains = len(self.bot.phishing_domains)
fields.append(
EmbedField(
@ -153,6 +154,7 @@ class UtilCog(Scale):
EmbedField(name="Position", value=str(role.position), inline=True),
EmbedField(name="Mentionable", value="Yes" if role.mentionable else "No", inline=True),
EmbedField(name="Member Count", value=str(len(role.members)), inline=True),
EmbedField(name="Created At", value=f"<t:{int(role.created_at.timestamp())}:F>"),
]
embed = build_embed(
title="",
@ -190,18 +192,15 @@ class UtilCog(Scale):
user_roles = user.roles
if user_roles:
user_roles = sorted(user.roles, key=lambda x: -x.position)
format_string = "%a, %b %-d, %Y %-I:%M %p"
if platform.system() == "Windows":
format_string = "%a, %b %#d, %Y %#I:%M %p"
fields = [
EmbedField(
name="Joined",
value=user.joined_at.strftime(format_string),
value=f"<t:{int(user.joined_at.timestamp())}:F>",
),
EmbedField(
name="Registered",
value=user.created_at.strftime(format_string),
value=f"<t:{int(user.created_at.timestamp())}:F>",
),
EmbedField(
name=f"Roles [{len(user_roles)}]",
@ -267,6 +266,7 @@ class UtilCog(Scale):
EmbedField(name="Threads", value=str(threads), inline=True),
EmbedField(name="Members", value=str(members), inline=True),
EmbedField(name="Roles", value=str(roles), inline=True),
EmbedField(name="Created At", value=f"<t:{int(guild.created_at.timestamp())}:F>"),
]
if len(role_list) < 1024:
fields.append(EmbedField(name="Role List", value=role_list, inline=False))
@ -375,6 +375,6 @@ class UtilCog(Scale):
await ctx.send(embed=embed, ephemeral=private)
def setup(bot: Snake) -> None:
def setup(bot: Client) -> None:
"""Add UtilCog to JARVIS"""
UtilCog(bot)

32
jarvis/cogs/verify.py
View file

@ -3,13 +3,13 @@ import asyncio
import logging
from random import randint
from dis_snek import InteractionContext, Scale, Snake
from dis_snek.models.discord.components import Button, ButtonStyles, spread_to_rows
from dis_snek.models.snek.application_commands import slash_command
from dis_snek.models.snek.command import cooldown
from dis_snek.models.snek.cooldowns import Buckets
from jarvis_core.db import q
from jarvis_core.db.models import Setting
from naff import Client, Cog, InteractionContext
from naff.models.discord.components import Button, ButtonStyles, spread_to_rows
from naff.models.naff.application_commands import slash_command
from naff.models.naff.command import cooldown
from naff.models.naff.cooldowns import Buckets
def create_layout() -> list:
@ -30,10 +30,10 @@ def create_layout() -> list:
return spread_to_rows(*buttons, max_in_row=3)
class VerifyCog(Scale):
class VerifyCog(Cog):
"""JARVIS Verify Cog."""
def __init__(self, bot: Snake):
def __init__(self, bot: Client):
self.bot = bot
self.logger = logging.getLogger(__name__)
@ -45,7 +45,13 @@ class VerifyCog(Scale):
if not role:
message = await ctx.send("This guild has not enabled verification", ephemeral=True)
return
if await ctx.guild.fetch_role(role.value) in ctx.author.roles:
verified_role = await ctx.guild.fetch_role(role.value)
if not verified_role:
await ctx.send("This guild has not enabled verification", ephemeral=True)
await role.delete()
return
if verified_role in ctx.author.roles:
await ctx.send("You are already verified.", ephemeral=True)
return
components = create_layout()
@ -69,12 +75,20 @@ class VerifyCog(Scale):
for component in row.components:
component.disabled = True
setting = await Setting.find_one(q(guild=ctx.guild.id, setting="verified"))
try:
role = await ctx.guild.fetch_role(setting.value)
await ctx.author.add_role(role, reason="Verification passed")
except AttributeError:
self.logger.warning("Verified role deleted before verification finished")
setting = await Setting.find_one(q(guild=ctx.guild.id, setting="unverified"))
if setting:
try:
role = await ctx.guild.fetch_role(setting.value)
await ctx.author.remove_role(role, reason="Verification passed")
except AttributeError:
self.logger.warning(
"Unverified role deleted before verification finished"
)
await response.context.edit_origin(
content=f"Welcome, {ctx.author.mention}. Please enjoy your stay.",
@ -94,6 +108,6 @@ class VerifyCog(Scale):
self.logger.debug(f"User {ctx.author.id} failed to verify before timeout")
def setup(bot: Snake) -> None:
def setup(bot: Client) -> None:
"""Add VerifyCog to JARVIS"""
VerifyCog(bot)

87
jarvis/config.py
View file

@ -1,96 +1,17 @@
"""Load the config for JARVIS"""
import os
from jarvis_core.config import Config as CConfig
from pymongo import MongoClient
from yaml import load
try:
from yaml import CLoader as Loader
except ImportError:
from yaml import Loader
class JarvisConfig(CConfig):
REQUIRED = ["token", "mongo", "urls"]
REQUIRED = ("token", "mongo", "urls", "redis")
OPTIONAL = {
"sync": False,
"log_level": "WARNING",
"scales": None,
"cogs": None,
"events": True,
"gitlab_token": None,
"max_messages": 1000,
"twitter": None,
"reddit": None,
"rook_token": None,
}
class Config(object):
"""Config singleton object for JARVIS"""
def __new__(cls, *args: list, **kwargs: dict):
"""Get the singleton config, or creates a new one."""
it = cls.__dict__.get("it")
if it is not None:
return it
cls.__it__ = it = object.__new__(cls)
it.init(*args, **kwargs)
return it
def init(
self,
token: str,
mongo: dict,
urls: dict,
sync: bool = False,
log_level: str = "WARNING",
cogs: list = None,
events: bool = True,
gitlab_token: str = None,
max_messages: int = 1000,
twitter: dict = None,
) -> None:
"""Initialize the config object."""
self.token = token
self.mongo = mongo
self.urls = urls
self.log_level = log_level
self.cogs = cogs
self.events = events
self.max_messages = max_messages
self.gitlab_token = gitlab_token
self.twitter = twitter
self.sync = sync or os.environ.get("SYNC_COMMANDS", False)
self.__db_loaded = False
self.__mongo = MongoClient(**self.mongo["connect"])
def get_db_config(self) -> None:
"""Load the database config objects."""
if not self.__db_loaded:
db = self.__mongo[self.mongo["database"]]
items = db.config.find()
for item in items:
setattr(self, item["key"], item["value"])
self.__db_loaded = True
@classmethod
def from_yaml(cls, y: dict) -> "Config":
"""Load the yaml config file."""
return cls(**y)
def get_config(path: str = "config.yaml") -> Config:
"""Get the config from the specified yaml file."""
if Config.__dict__.get("it"):
return Config()
with open(path) as f:
raw = f.read()
y = load(raw, Loader=Loader)
config = Config.from_yaml(y)
config.get_db_config()
return config
def reload_config() -> None:
"""Force reload of the config singleton on next call."""
if "it" in Config.__dict__:
Config.__dict__.pop("it")

7
jarvis/const.py Normal file
View file

@ -0,0 +1,7 @@
"""JARVIS constants."""
from importlib.metadata import version as _v
try:
__version__ = _v("jarvis")
except Exception:
__version__ = "0.0.0"

52
jarvis/data/robotcamo.py
View file

@ -50,32 +50,32 @@ emotes = {
}
names = {
852317928572715038: "rcA",
852317954975727679: "rcB",
852317972424818688: "rcC",
852317990238421003: "rcD",
852318044503539732: "rcE",
852318058353786880: "rcF",
852318073994477579: "rcG",
852318105832259614: "rcH",
852318122278125580: "rcI",
852318145074167818: "rcJ",
852318159952412732: "rcK",
852318179358408704: "rcL",
852318241555873832: "rcM",
852318311115128882: "rcN",
852318329951223848: "rcO",
852318344643477535: "rcP",
852318358920757248: "rcQ",
852318385638211594: "rcR",
852318401166311504: "rcS",
852318421524938773: "rcT",
852318435181854742: "rcU",
852318453204647956: "rcV",
852318470267731978: "rcW",
852318484749877278: "rcX",
852318504564555796: "rcY",
852318519449092176: "rcZ",
852317928572715038: "rcLetterA",
852317954975727679: "rcLetterB",
852317972424818688: "rcLetterC",
852317990238421003: "rcLetterD",
852318044503539732: "rcLetterE",
852318058353786880: "rcLetterF",
852318073994477579: "rcLetterG",
852318105832259614: "rcLetterH",
852318122278125580: "rcLetterI",
852318145074167818: "rcLetterJ",
852318159952412732: "rcLetterK",
852318179358408704: "rcLetterL",
852318241555873832: "rcLetterM",
852318311115128882: "rcLetterN",
852318329951223848: "rcLetterO",
852318344643477535: "rcLetterP",
852318358920757248: "rcLetterQ",
852318385638211594: "rcLetterR",
852318401166311504: "rcLetterS",
852318421524938773: "rcLetterT",
852318435181854742: "rcLetterU",
852318453204647956: "rcLetterV",
852318470267731978: "rcLetterW",
852318484749877278: "rcLetterX",
852318504564555796: "rcLetterY",
852318519449092176: "rcLetterZ",
860663352740151316: "rc1",
860662785243348992: "rc2",
860662950011469854: "rc3",

17
jarvis/utils/__init__.py
View file

@ -3,14 +3,11 @@ from datetime import datetime, timezone
from pkgutil import iter_modules
import git
from dis_snek.models.discord.embed import Embed, EmbedField
from dis_snek.models.discord.guild import AuditLogEntry
from dis_snek.models.discord.user import Member
from naff.models.discord.embed import Embed, EmbedField
from naff.models.discord.guild import AuditLogEntry
from naff.models.discord.user import Member
import jarvis.cogs
from jarvis.config import get_config
__all__ = ["cachecog", "permissions"]
from jarvis.config import JarvisConfig
def build_embed(
@ -64,11 +61,11 @@ def modlog_embed(
return embed
def get_extensions(path: str = jarvis.cogs.__path__) -> list:
def get_extensions(path: str) -> list:
"""Get JARVIS cogs."""
config = get_config()
config = JarvisConfig.from_yaml()
vals = config.cogs or [x.name for x in iter_modules(path)]
return ["jarvis.cogs.{}".format(x) for x in vals]
return [f"jarvis.cogs.{x}" for x in vals]
def update() -> int:

133
jarvis/utils/cogs.py
View file

@ -1,64 +1,28 @@
"""Cog wrapper for command caching."""
from datetime import datetime, timedelta, timezone
import logging
from datetime import timedelta
from dis_snek import InteractionContext, Scale, Snake
from dis_snek.client.utils.misc_utils import find
from dis_snek.models.discord.embed import EmbedField
from dis_snek.models.snek.tasks.task import Task
from dis_snek.models.snek.tasks.triggers import IntervalTrigger
from jarvis_core.db import q
from jarvis_core.db.models import (
Action,
Ban,
Kick,
Modlog,
Mute,
Note,
Setting,
Warning,
)
from jarvis_core.db.models import Action, Ban, Kick, Modlog, Mute, Setting, Warning
from naff import Client, Cog, InteractionContext
from naff.models.discord.components import ActionRow, Button, ButtonStyles
from naff.models.discord.embed import EmbedField
from jarvis.utils import build_embed
MODLOG_LOOKUP = {"Ban": Ban, "Kick": Kick, "Mute": Mute, "Warning": Warning}
IGNORE_COMMANDS = {"Ban": ["bans"], "Kick": [], "Mute": ["unmute"], "Warning": ["warnings"]}
class CacheCog(Scale):
"""Cog wrapper for command caching."""
def __init__(self, bot: Snake):
self.bot = bot
self.cache = {}
self._expire_interaction.start()
def check_cache(self, ctx: InteractionContext, **kwargs: dict) -> dict:
"""Check the cache."""
if not kwargs:
kwargs = {}
return find(
lambda x: x["command"] == ctx.subcommand_name # noqa: W503
and x["user"] == ctx.author.id # noqa: W503
and x["guild"] == ctx.guild.id # noqa: W503
and all(x[k] == v for k, v in kwargs.items()), # noqa: W503
self.cache.values(),
)
@Task.create(IntervalTrigger(minutes=1))
async def _expire_interaction(self) -> None:
keys = list(self.cache.keys())
for key in keys:
if self.cache[key]["timeout"] <= datetime.now(tz=timezone.utc) + timedelta(minutes=1):
del self.cache[key]
class ModcaseCog(Scale):
class ModcaseCog(Cog):
"""Cog wrapper for moderation case logging."""
def __init__(self, bot: Snake):
def __init__(self, bot: Client):
self.bot = bot
self.add_scale_postrun(self.log)
self.logger = logging.getLogger(__name__)
self.add_cog_postrun(self.log)
async def log(self, ctx: InteractionContext, *args: list, **kwargs: dict) -> None:
async def log(self, ctx: InteractionContext, *_args: list, **kwargs: dict) -> None:
"""
Log a moderation activity in a moderation case.
@ -67,34 +31,33 @@ class ModcaseCog(Scale):
"""
name = self.__name__.replace("Cog", "")
if name not in ["Lock", "Lockdown", "Purge", "Roleping"]:
if name in MODLOG_LOOKUP and ctx.invoke_target not in IGNORE_COMMANDS[name]:
user = kwargs.pop("user", None)
if not user and not ctx.target_id:
self.logger.warn(f"Admin action {name} missing user, exiting")
self.logger.warning("Admin action %s missing user, exiting", name)
return
elif ctx.target_id:
if ctx.target_id:
user = ctx.target
coll = MODLOG_LOOKUP.get(name, None)
if not coll:
self.logger.warn(f"Unsupported action {name}, exiting")
self.logger.warning("Unsupported action %s, exiting", name)
return
action = await coll.find_one(q(user=user.id, guild=ctx.guild_id, active=True))
action = await coll.find_one(
q(user=user.id, guild=ctx.guild_id, active=True), sort=[("_id", -1)]
)
if not action:
self.logger.warn(f"Missing action {name}, exiting")
self.logger.warning("Missing action %s, exiting", name)
return
action = Action(action_type=name.lower(), parent=action.id)
note = Note(admin=self.bot.user.id, content="Moderation case opened automatically")
await Modlog(user=user.id, admin=ctx.author.id, actions=[action], notes=[note]).commit()
notify = await Setting.find_one(q(guild=ctx.guild.id, setting="notify", value=True))
if notify and name not in ["Kick", "Ban"]: # Ignore Kick and Ban, as these are unique
fields = [
if notify and name not in ("Kick", "Ban"): # Ignore Kick and Ban, as these are unique
fields = (
EmbedField(name="Action Type", value=name, inline=False),
EmbedField(
name="Reason", value=kwargs.get("reason", None) or "N/A", inline=False
),
]
)
embed = build_embed(
title="Admin action taken",
description=f"Admin action has been taken against you in {ctx.guild.name}",
@ -103,4 +66,54 @@ class ModcaseCog(Scale):
guild_url = f"https://discord.com/channels/{ctx.guild.id}"
embed.set_author(name=ctx.guild.name, icon_url=ctx.guild.icon.url, url=guild_url)
embed.set_thumbnail(url=ctx.guild.icon.url)
try:
await user.send(embed=embed)
except Exception:
self.logger.debug("User not warned of action due to closed DMs")
modlog = await Modlog.find_one(q(user=user.id, guild=ctx.guild.id, open=True))
if modlog:
m_action = Action(action_type=name.lower(), parent=action.id)
modlog.actions.append(m_action)
await modlog.commit()
return
lookup_key = f"{user.id}|{ctx.guild.id}"
async with self.bot.redis.lock("lock|" + lookup_key):
if await self.bot.redis.get(lookup_key):
self.logger.debug(f"User {user.id} in {ctx.guild.id} already has pending case")
return
modlog = await Setting.find_one(q(guild=ctx.guild.id, setting="modlog"))
if not modlog:
return
channel = await ctx.guild.fetch_channel(modlog.value)
if not channel:
self.logger.warn(
f"Guild {ctx.guild.id} modlog channel no longer exists, deleting"
)
await modlog.delete()
return
embed = build_embed(
title="Recent Action Taken",
description=f"Would you like to open a moderation case for {user.mention}?",
fields=[],
)
embed.set_author(
name=user.username + "#" + user.discriminator, icon_url=user.display_avatar.url
)
components = [
ActionRow(
Button(style=ButtonStyles.RED, emoji="✖️", custom_id="modcase|no"),
Button(style=ButtonStyles.GREEN, emoji="✔️", custom_id="modcase|yes"),
)
]
message = await channel.send(embed=embed, components=components)
await self.bot.redis.set(
lookup_key, f"{name.lower()}|{action.id}", ex=timedelta(days=7)
)
await self.bot.redis.set(f"msg|{message.id}", user.id, ex=timedelta(days=7))

9
jarvis/utils/embeds.py
View file

@ -1,7 +1,8 @@
"""JARVIS bot-specific embeds."""
from dis_snek.models.discord.embed import Embed, EmbedField
from dis_snek.models.discord.user import Member
from jarvis_core.util import build_embed
from naff.models.discord.embed import Embed, EmbedField
from naff.models.discord.user import Member
from jarvis.utils import build_embed
def warning_embed(user: Member, reason: str) -> Embed:
@ -12,7 +13,7 @@ def warning_embed(user: Member, reason: str) -> Embed:
user: User to warn
reason: Warning reason
"""
fields = [EmbedField(name="Reason", value=reason, inline=False)]
fields = (EmbedField(name="Reason", value=reason, inline=False),)
embed = build_embed(
title="Warning", description=f"{user.mention} has been warned", fields=fields
)

9
jarvis/utils/permissions.py
View file

@ -1,7 +1,7 @@
"""Permissions wrappers."""
from dis_snek import InteractionContext, Permissions
from naff import InteractionContext, Permissions
from jarvis.config import get_config
from jarvis.config import JarvisConfig
def user_is_bot_admin() -> bool:
@ -9,8 +9,9 @@ def user_is_bot_admin() -> bool:
async def predicate(ctx: InteractionContext) -> bool:
"""Command check predicate."""
if getattr(get_config(), "admins", None):
return ctx.author.id in get_config().admins
cfg = JarvisConfig.from_yaml()
if getattr(cfg, "admins", None):
return ctx.author.id in cfg.admins
else:
return False

217
jarvis/utils/updates.py Normal file
View file

@ -0,0 +1,217 @@
"""JARVIS update handler."""
import asyncio
import logging
from dataclasses import dataclass
from importlib import import_module
from inspect import getmembers, isclass
from pkgutil import iter_modules
from types import FunctionType, ModuleType
from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional
import git
from naff.client.errors import ExtensionNotFound
from naff.client.utils.misc_utils import find, find_all
from naff.models.naff.application_commands import SlashCommand
from naff.models.naff.cog import Cog
from rich.table import Table
import jarvis.cogs
if TYPE_CHECKING:
from naff.client.client import Client
_logger = logging.getLogger(__name__)
@dataclass
class UpdateResult:
"""JARVIS update result."""
old_hash: str
new_hash: str
table: Table
added: List[str]
removed: List[str]
changed: List[str]
lines: Dict[str, int]
def get_all_commands(module: ModuleType = jarvis.cogs) -> Dict[str, Callable]:
"""Get all SlashCommands from a specified module."""
commands = {}
def validate_ires(entry: Any) -> bool:
return isclass(entry) and issubclass(entry, Cog) and entry is not Cog
def validate_cog(cog: FunctionType) -> bool:
return isinstance(cog, SlashCommand)
for item in iter_modules(module.__path__):
new_module = import_module(f"{module.__name__}.{item.name}")
if item.ispkg:
if cmds := get_all_commands(new_module):
commands.update(cmds)
else:
inspect_result = getmembers(new_module)
cogs = []
for _, val in inspect_result:
if validate_ires(val):
cogs.append(val)
for cog in cogs:
values = cog.__dict__.values()
commands[cog.__module__] = find_all(lambda x: isinstance(x, SlashCommand), values)
return {k: v for k, v in commands.items() if v}
def get_git_changes(repo: git.Repo) -> dict:
"""Get all Git changes"""
logger = _logger
logger.debug("Getting all git changes")
current_hash = repo.head.ref.object.hexsha
tracking = repo.head.ref.tracking_branch()
file_changes = {}
for commit in tracking.commit.iter_items(repo, f"{repo.head.ref.path}..{tracking.path}"):
if commit.hexsha == current_hash:
break
files = commit.stats.files
file_changes.update(
{key: {"insertions": 0, "deletions": 0, "lines": 0} for key in files.keys()}
)
for file, stats in files.items():
if file not in file_changes:
file_changes[file] = {"insertions": 0, "deletions": 0, "lines": 0}
for key, val in stats.items():
file_changes[file][key] += val
logger.debug("Found %i changed files", len(file_changes))
table = Table(title="File Changes")
table.add_column("File", justify="left", style="white", no_wrap=True)
table.add_column("Insertions", justify="center", style="green")
table.add_column("Deletions", justify="center", style="red")
table.add_column("Lines", justify="center", style="magenta")
i_total = 0
d_total = 0
l_total = 0
for file, stats in file_changes.items():
i_total += stats["insertions"]
d_total += stats["deletions"]
l_total += stats["lines"]
table.add_row(
file,
str(stats["insertions"]),
str(stats["deletions"]),
str(stats["lines"]),
)
logger.debug("%i insertions, %i deletions, %i total", i_total, d_total, l_total)
table.add_row("Total", str(i_total), str(d_total), str(l_total))
return {
"table": table,
"lines": {"inserted_lines": i_total, "deleted_lines": d_total, "total_lines": l_total},
}
async def update(bot: "Client") -> Optional[UpdateResult]:
"""
Update JARVIS and return an UpdateResult.
Args:
bot: Bot instance
Returns:
UpdateResult object
"""
logger = _logger
repo = git.Repo(".")
current_hash = repo.head.object.hexsha
origin = repo.remotes.origin
origin.fetch()
remote_hash = origin.refs[repo.active_branch.name].object.hexsha
if current_hash != remote_hash:
logger.info("Updating from %s to %s", current_hash, remote_hash)
current_commands = get_all_commands()
changes = get_git_changes(repo)
origin.pull()
await asyncio.sleep(3)
new_commands = get_all_commands()
logger.info("Checking if any modules need reloaded...")
reloaded = []
loaded = []
unloaded = []
logger.debug("Checking for removed cogs")
for module in current_commands.keys():
if module not in new_commands:
logger.debug("Module %s removed after update", module)
bot.drop_cog(module)
unloaded.append(module)
logger.debug("Checking for new/modified commands")
for module, commands in new_commands.items():
logger.debug("Processing %s", module)
if module not in current_commands:
bot.load_cog(module)
loaded.append(module)
elif len(current_commands[module]) != len(commands):
try:
bot.reload_cog(module)
except ExtensionNotFound:
bot.load_cog(module)
reloaded.append(module)
else:
for command in commands:
old_command = find(
lambda x: x.resolved_name == command.resolved_name, current_commands[module]
)
# Extract useful info
old_args = old_command.options
if old_args:
old_arg_names = [x.name for x in old_args]
new_args = command.options
if new_args:
new_arg_names = [x.name for x in new_args]
# No changes
if not old_args and not new_args:
continue
# Check if number arguments have changed
if len(old_args) != len(new_args):
try:
bot.reload_cog(module)
except ExtensionNotFound:
bot.load_cog(module)
reloaded.append(module)
elif any(x not in old_arg_names for x in new_arg_names) or any(
x not in new_arg_names for x in old_arg_names
):
try:
bot.reload_cog(module)
except ExtensionNotFound:
bot.load_cog(module)
reloaded.append(module)
elif any(new_args[idx].type != x.type for idx, x in enumerate(old_args)):
try:
bot.reload_cog(module)
except ExtensionNotFound:
bot.load_cog(module)
reloaded.append(module)
return UpdateResult(
old_hash=current_hash,
new_hash=remote_hash,
added=loaded,
removed=unloaded,
changed=reloaded,
**changes,
)
return None

762
poetry.lock generated
View file

File diff suppressed because it is too large Load diff

9
pyproject.toml
View file

@ -1,13 +1,12 @@
[tool.poetry]
name = "jarvis"
version = "2.0.0b1"
version = "2.0.0b2"
description = "J.A.R.V.I.S. admin bot"
authors = ["Zevaryx <zevaryx@gmail.com>"]
[tool.poetry.dependencies]
python = "^3.10"
PyYAML = "^6.0"
dis-snek = "*"
GitPython = "^3.1.26"
mongoengine = "^0.23.1"
opencv-python = "^4.5.5"
@ -22,7 +21,11 @@ aiohttp = "^3.8.1"
pastypy = "^1.0.1"
dateparser = "^1.1.1"
aiofile = "^3.7.4"
molter = "^0.11.0"
asyncpraw = "^7.5.0"
rook = "^0.1.170"
rich = "^12.3.0"
jurigged = "^0.5.0"
aioredis = "^2.0.1"
[build-system]
requires = ["poetry-core>=1.0.0"]