feat(core): Implement event log interface.

This commit is contained in:
2023-10-14 01:07:46 +03:00
parent 4457e60120
commit 7b6290b73e
4 changed files with 237 additions and 9 deletions

View File

@@ -1,6 +1,3 @@
from .cog import CoreCog
from .config import ConfigCog
from babel.translator import LocalBabel
@@ -8,5 +5,8 @@ babel = LocalBabel('lion-core')
async def setup(bot):
from .cog import CoreCog
from .config import ConfigCog
await bot.add_cog(CoreCog(bot))
await bot.add_cog(ConfigCog(bot))

View File

@@ -1,20 +1,75 @@
from typing import Optional, TYPE_CHECKING
from enum import Enum
import asyncio
import datetime as dt
import pytz
import discord
import logging
from meta import LionBot
from utils.lib import Timezoned
from meta import LionBot, conf
from meta.logger import log_wrap
from utils.lib import Timezoned, utc_now
from settings.groups import ModelConfig, SettingDotDict
from babel.translator import ctx_locale
from .hooks import HookedChannel
from .data import CoreData
from . import babel
if TYPE_CHECKING:
# TODO: Import Settings for Config type hinting
pass
_p = babel._p
logger = logging.getLogger(__name__)
event_fields = {
'start': (
_p('eventlog|field:start|name', "Start"),
"{value}",
True,
),
'expiry': (
_p('eventlog|field:expiry|name', "Expires"),
"{value}",
True,
),
'roles_given' : (
_p('eventlog|field:roles_given|name', "Roles Given"),
"{value}",
True,
),
'roles_taken' : (
_p('eventlog|field:roles_given|name', "Roles Taken"),
"{value}",
True,
),
'coins_earned' : (
_p('eventlog|field:coins_earned|name', "Coins Earned"),
"{coin} {{value}}".format(coin=conf.emojis.coin),
True,
),
'price' : (
_p('eventlog|field:price|name', "Price"),
"{coin} {{value}}".format(coin=conf.emojis.coin),
True,
),
'memberid': (
_p('eventlog|field:memberid|name', "Member"),
"<@{value}>",
True,
),
'channelid': (
_p('eventlog|field:channelid|name', "Channel"),
"<#{value}>",
True
),
}
class VoiceMode(Enum):
STUDY = 0
VOICE = 1
@@ -49,7 +104,16 @@ class LionGuild(Timezoned):
No guarantee is made that the client is in the corresponding Guild,
or that the corresponding Guild even exists.
"""
__slots__ = ('bot', 'data', 'guildid', 'config', '_guild', 'voice_lock', '__weakref__')
__slots__ = (
'bot', 'data',
'guildid',
'config',
'_guild',
'voice_lock',
'_eventlogger',
'_tasks',
'__weakref__'
)
Config = GuildConfig
settings = Config.settings
@@ -68,6 +132,24 @@ class LionGuild(Timezoned):
# Avoids voice race-states
self.voice_lock = asyncio.Lock()
# HookedChannel managing the webhook used to send guild event logs
# May be None if no event log is set or if the channel does not exist
self._eventlogger: Optional[HookedChannel] = None
# Set of background tasks associated with this guild (e.g. event logs)
# In theory we should ensure these are finished before the lguild is gcd
# But this is *probably* not an actual problem in practice
self._tasks = set()
@property
def eventlogger(self) -> Optional[HookedChannel]:
channelid = self.data.event_log_channel
if channelid is None:
self._eventlogger = None
elif self._eventlogger is None or self._eventlogger.channelid != channelid:
self._eventlogger = self.bot.core.hooked_channel(channelid)
return self._eventlogger
@property
def guild(self):
if self._guild is None:
@@ -93,4 +175,149 @@ class LionGuild(Timezoned):
"""
if self.data.name != guild.name:
await self.data.update(name=guild.name)
@log_wrap(action='get event hook')
async def get_event_hook(self) -> Optional[discord.Webhook]:
hooked = self.eventlogger
ctx_locale.set(self.locale)
if hooked:
hook = await hooked.get_webhook()
if hook is not None:
pass
elif (channel := hooked.channel) is None:
# Event log channel doesn't exist
pass
elif not channel.permissions_for(channel.guild.me).manage_webhooks:
# Cannot create a webhook here
if channel.permissions_for(channel.guild.me).send_messages:
t = self.bot.translator.t
try:
await channel.send(t(_p(
'eventlog|error:manage_webhooks',
"This channel is configured as an event log, "
"but I am missing the 'Manage Webhooks' permission here."
)))
except discord.HTTPException:
pass
else:
# We should be able to create the hook
t = self.bot.translator.t
try:
hook = await hooked.create_webhook(
name=t(_p(
'eventlog|create|name',
"{bot_name} Event Log"
)).format(bot_name=channel.guild.me.name),
reason=t(_p(
'eventlog|create|audit_reason',
"Creating event log webhook"
)),
)
except discord.HTTPException:
logger.warning(
f"Unexpected exception while creating event log webhook for <gid: {self.guildid}>",
exc_info=True
)
return hook
@log_wrap(action="Log Event")
async def _log_event(self, embed: discord.Embed, retry=True):
logger.debug(f"Logging event log event: {embed.to_dict()}")
hook = await self.get_event_hook()
if hook is not None:
try:
await hook.send(embed=embed)
except discord.NotFound:
logger.info(
f"Event log in <gid: {self.guildid}> invalidated. Recreating: {retry}"
)
hooked = self.eventlogger
if hooked is not None:
await hooked.invalidate(hook)
if retry:
await self._log_event(embed, retry=False)
except discord.HTTPException:
logger.warning(
f"Discord exception occurred sending event log event: {embed.to_dict()}.",
exc_info=True
)
except Exception:
logger.exception(
f"Unknown exception occurred sending event log event: {embed.to_dict()}."
)
def log_event(self,
title: Optional[str]=None, description: Optional[str]=None,
timestamp: Optional[dt.datetime]=None,
*,
embed: Optional[discord.Embed] = None,
fields: dict[str, tuple[str, bool]]={},
**kwargs: str | int):
"""
Synchronously log an event to the guild event log.
Does nothing if the event log has not been set up.
Parameters
----------
title: str
Embed title
description: str
Embed description
timestamp: dt.datetime
Embed timestamp. Defaults to `now` if not given.
embed: discord.Embed
Optional base embed to use.
May be used to completely customise log message.
fields: dict[str, tuple[str, bool]]
Optional embed fields to add.
kwargs: str | int
Optional embed fields to add to the embed.
These differ from `fields` in that the kwargs keys will be automatically matched and localised
if possible.
These will be added before the `fields` given.
"""
t = self.bot.translator.t
# Build embed
base = embed if embed is not None else discord.Embed(colour=discord.Colour.dark_orange())
if description is not None:
base.description = description
if title is not None:
base.title = title
if timestamp is not None:
base.timestamp = timestamp
else:
base.timestamp = utc_now()
# Add embed fields
for key, value in kwargs.items():
if value is None:
continue
if key in event_fields:
_field_name, _field_value, inline = event_fields[key]
field_name = t(_field_name, locale=self.locale)
field_value = _field_value.format(value=value)
else:
field_name = key
field_value = value
inline = False
base.add_field(
name=field_name,
value=field_value,
inline=inline
)
for key, (value, inline) in fields.items():
base.add_field(
name=key,
value=value,
inline=inline,
)
# Send embed
task = asyncio.create_task(self._log_event(embed=base), name='event-log')
self._tasks.add(task)
task.add_done_callback(self._tasks.discard)

View File

@@ -22,6 +22,7 @@ from modules.statistics.settings import StatisticsDashboard
from modules.member_admin.settingui import MemberAdminDashboard
from modules.moderation.settingui import ModerationDashboard
from modules.video_channels.settingui import VideoDashboard
from modules.config.settingui import GeneralDashboard
from . import babel, logger
@@ -35,7 +36,7 @@ class GuildDashboard(BasePager):
Paged UI providing an overview of the guild configuration.
"""
pages = [
(MemberAdminDashboard, LocaleDashboard, EconomyDashboard,),
(MemberAdminDashboard, LocaleDashboard, EconomyDashboard, GeneralDashboard,),
(ModerationDashboard, VideoDashboard,),
(VoiceTrackerDashboard, TextTrackerDashboard, RankDashboard, StatisticsDashboard,),
(TasklistDashboard, RoomDashboard, TimerDashboard,),

View File

@@ -95,7 +95,7 @@ class GeneralSettingUI(ConfigUI):
class GeneralDashboard(DashboardSection):
section_name = _p(
"dash:general|title",
"General Dashboard Settings ({commands[configure general]})"
"General Configuration ({commands[configure general]})"
)
_option_name = _p(
"dash:general|option|name",