155 lines
5.1 KiB
Python
155 lines
5.1 KiB
Python
from data.queries import ORDER
|
|
import discord
|
|
from discord.ext import commands as cmds
|
|
from discord import app_commands as appcmds
|
|
|
|
from meta import LionBot, LionCog, LionContext
|
|
from meta.logger import log_wrap
|
|
from utils.lib import utc_now
|
|
|
|
from . import logger
|
|
from ..data import VoiceLogData, VoiceLogGuild, VoiceLogSession
|
|
from .lib import ThreadedWebhook
|
|
|
|
|
|
class VoiceLogCog(LionCog):
|
|
def __init__(self, bot: LionBot):
|
|
self.bot = bot
|
|
self.data = bot.db.load_registry(VoiceLogData())
|
|
|
|
async def cog_load(self):
|
|
await self.data.init()
|
|
|
|
@LionCog.listener("on_voice_state_update")
|
|
@log_wrap(action="Voice Log Update")
|
|
async def voicelog_update(
|
|
self,
|
|
member: discord.Member,
|
|
before: discord.VoiceState,
|
|
after: discord.VoiceState,
|
|
):
|
|
if member.bot:
|
|
return
|
|
|
|
after_channel = after.channel
|
|
before_channel = before.channel
|
|
if after_channel == before.channel:
|
|
return
|
|
|
|
row = await VoiceLogGuild.fetch(member.guild.id)
|
|
if row is None or row.webhook_url is None:
|
|
return
|
|
|
|
hook = ThreadedWebhook.from_url(row.webhook_url, client=self.bot)
|
|
|
|
now = utc_now()
|
|
embed = discord.Embed(timestamp=now)
|
|
|
|
if before_channel is None:
|
|
embed.title = "User Joined Voice Channel"
|
|
embed.description = f"{member.mention} joined {after_channel.mention}"
|
|
elif after_channel is None:
|
|
embed.title = "User Left Voice Channel"
|
|
embed.description = f"{member.mention} left {before_channel.mention}"
|
|
else:
|
|
embed.title = "User Switched Voice Channel"
|
|
embed.description = f"{member.mention} moved from {before_channel.mention} to {after_channel.mention}"
|
|
|
|
session = None
|
|
if before_channel:
|
|
session = await self.end_voice_session(member, before_channel, now)
|
|
if not session:
|
|
embed.add_field(
|
|
name="Warning",
|
|
value="Could not find voice session to end, statistics may be incorrect.",
|
|
)
|
|
else:
|
|
embed.add_field(
|
|
name="Session",
|
|
value=f"Voice session lasted {session.duration} seconds!",
|
|
)
|
|
if after_channel:
|
|
session = await self.start_voice_session(member, after_channel, now)
|
|
|
|
if session:
|
|
embed.set_footer(text=f"Session #{session.sessionid}")
|
|
|
|
await hook.send(embed=embed)
|
|
|
|
async def start_voice_session(self, member, channel, started_at):
|
|
session = await VoiceLogSession.create(
|
|
guildid=member.guild.id,
|
|
userid=member.id,
|
|
channelid=channel.id,
|
|
joined_at=started_at,
|
|
)
|
|
return session
|
|
|
|
async def end_voice_session(self, member, channel, ended_at):
|
|
# Get the last open voice session to close it
|
|
open_sessions = await VoiceLogSession.fetch_where(
|
|
guildid=member.guild_id,
|
|
userid=member.id,
|
|
channelid=channel.id,
|
|
duration=None,
|
|
).order_by("joined_at", ORDER.DESC)
|
|
if not open_sessions:
|
|
session = None
|
|
else:
|
|
session = open_sessions[0]
|
|
await session.update(
|
|
duration=int((ended_at - session.joined_at).total_seconds())
|
|
)
|
|
|
|
return session
|
|
|
|
@cmds.hybrid_group(
|
|
name="voicelog", description="Base command group for the voice logging system"
|
|
)
|
|
@appcmds.default_permissions(manage_guild=True)
|
|
async def voicelog_group(self, ctx: LionContext): ...
|
|
|
|
@voicelog_group.command(
|
|
name="enable",
|
|
description="Enable voice activity logging and set the webhook url to use.",
|
|
)
|
|
async def voicelog_enable(self, ctx: LionContext, webhook_url: str):
|
|
if not ctx.guild:
|
|
return
|
|
if not ctx.author.guild_permissions.manage_guild:
|
|
return
|
|
|
|
webhook = ThreadedWebhook.from_url(webhook_url, client=self.bot)
|
|
try:
|
|
embed = discord.Embed(
|
|
title="Testing",
|
|
description="Testing logging webhook, feel free to delete.",
|
|
)
|
|
await webhook.send(embed=embed, wait=True)
|
|
existing = await VoiceLogGuild.fetch(ctx.guild.id)
|
|
if existing:
|
|
await existing.update(webhook_url=webhook_url)
|
|
else:
|
|
await VoiceLogGuild.create(
|
|
guildid=ctx.guild.id, webhook_url=webhook_url
|
|
)
|
|
|
|
await ctx.reply("Voice logging enabled!")
|
|
except discord.HTTPException:
|
|
await ctx.error_reply("Could not post to the given webhook!")
|
|
|
|
@voicelog_group.command(
|
|
name="disable", description="Disable voice activity logging in this server."
|
|
)
|
|
async def voicelog_disable(self, ctx: LionContext):
|
|
if not ctx.guild:
|
|
return
|
|
if not ctx.author.guild_permissions.manage_guild:
|
|
return
|
|
|
|
await self.data.voicelog_guilds.update_where(guildid=ctx.guild.id).set(
|
|
webhook_url=None
|
|
)
|
|
|
|
await ctx.reply("Voice Activity logging disabled.")
|