|
|
|
@@ -0,0 +1,260 @@
|
|
|
|
|
"""
|
|
|
|
|
discord channel logger
|
|
|
|
|
|
|
|
|
|
leveled logging to a discord channel via embeds. attach to your bot (e.g.
|
|
|
|
|
bot.log) and call bot.log.info(...), bot.log.critical(...), etc. config-free:
|
|
|
|
|
static identity is injected at construction; per-guild channel routing is
|
|
|
|
|
read live from bot.settings at call time (no restart needed).
|
|
|
|
|
|
|
|
|
|
from dpy_logger import DPYLogger
|
|
|
|
|
|
|
|
|
|
bot.log = DPYLogger(
|
|
|
|
|
bot, guild_id, channel_id,
|
|
|
|
|
colors=log_colors, # optional, falls back to defaults
|
|
|
|
|
pings=authorized_devs, # mentioned on critical()
|
|
|
|
|
timezone=tz,
|
|
|
|
|
footer=bot_footer,
|
|
|
|
|
avatar=bot_avatar,
|
|
|
|
|
)
|
|
|
|
|
await bot.log.initialize() # resolves ids -> objects
|
|
|
|
|
await bot.log.success("user promoted", action="promote", actor=ctx.author)
|
|
|
|
|
|
|
|
|
|
levels: debug, info, success, fail (alias failure), task, critical. a per-call
|
|
|
|
|
guild= argument routes to that guild's bot.settings[guild.id]['channels']['logs']
|
|
|
|
|
instead of the construction channel. pass embed_builder=fn(logger, level, action,
|
|
|
|
|
actor, details) -> discord.Embed to restyle without subclassing, or override
|
|
|
|
|
build_embed in a subclass for complex cases. no feed/announcement method by
|
|
|
|
|
design; subclass and reuse _resolve for project-specific log types.
|
|
|
|
|
|
|
|
|
|
dual sink: every call also mirrors to the stdlib logger (getLogger(__name__))
|
|
|
|
|
before the discord send, so the record survives even if discord fails. set
|
|
|
|
|
log_to_file=False at construction, or per call, to disable.
|
|
|
|
|
|
|
|
|
|
errors: setup raises, sends swallow. resolution failures (unresolvable
|
|
|
|
|
guild/channel, bad config, a non-text channel, an invalid timezone) raise
|
|
|
|
|
ValueError from `initialize()`, with underlying discord exceptions normalized
|
|
|
|
|
to it. per-call send failures never propagate - they fall back to the stdlib
|
|
|
|
|
logger so a transient discord failure never breaks the caller's command.
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
|
|
import logging
|
|
|
|
|
import discord
|
|
|
|
|
from datetime import datetime, tzinfo
|
|
|
|
|
from typing import Callable, Optional
|
|
|
|
|
|
|
|
|
|
_log = logging.getLogger(__name__)
|
|
|
|
|
|
|
|
|
|
DEFAULT_COLORS = {
|
|
|
|
|
"debug": 0x808080,
|
|
|
|
|
"info": 0x709288,
|
|
|
|
|
"success": 0x00E27D,
|
|
|
|
|
"fail": 0xD93025,
|
|
|
|
|
"task": 0x5AC6E1,
|
|
|
|
|
"critical": 0x000000,
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
DISCORD_CONTENT_LIMIT = 2000
|
|
|
|
|
DISCORD_FIELD_VALUE_LIMIT = 1024
|
|
|
|
|
|
|
|
|
|
LEVEL_MAP = {
|
|
|
|
|
"debug": logging.DEBUG,
|
|
|
|
|
"info": logging.INFO,
|
|
|
|
|
"success": logging.INFO,
|
|
|
|
|
"task": logging.INFO,
|
|
|
|
|
"fail": logging.ERROR,
|
|
|
|
|
"critical": logging.CRITICAL,
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
class DPYLogger:
|
|
|
|
|
"""leveled discord channel logger; attach to the bot as bot.log"""
|
|
|
|
|
|
|
|
|
|
def __init__(
|
|
|
|
|
self,
|
|
|
|
|
bot,
|
|
|
|
|
guild=None,
|
|
|
|
|
channel=None,
|
|
|
|
|
*,
|
|
|
|
|
colors: Optional[dict] = None,
|
|
|
|
|
pings: Optional[list] = None,
|
|
|
|
|
timezone: Optional[object] = None,
|
|
|
|
|
footer: str = "",
|
|
|
|
|
avatar: Optional[str] = None,
|
|
|
|
|
alert_here: bool = False,
|
|
|
|
|
embed_builder: Optional[Callable] = None,
|
|
|
|
|
log_to_file: bool = True,
|
|
|
|
|
):
|
|
|
|
|
self.bot = bot
|
|
|
|
|
self.guild = guild
|
|
|
|
|
self.channel = channel
|
|
|
|
|
self.pings = pings or []
|
|
|
|
|
self.timezone = timezone
|
|
|
|
|
self.footer = footer
|
|
|
|
|
self.avatar = avatar
|
|
|
|
|
self.alert_here = alert_here
|
|
|
|
|
self.colors = {**DEFAULT_COLORS, **(colors or {})}
|
|
|
|
|
self._embed_builder = embed_builder
|
|
|
|
|
self.log_to_file = log_to_file
|
|
|
|
|
|
|
|
|
|
async def initialize(self):
|
|
|
|
|
"""resolve guild/channel from ids to objects; call once after construction"""
|
|
|
|
|
if self.timezone is not None and not isinstance(self.timezone, tzinfo):
|
|
|
|
|
raise ValueError(f"[dpy_logger] timezone must be a tzinfo instance, not {self.timezone!r}")
|
|
|
|
|
if isinstance(self.guild, int):
|
|
|
|
|
self.guild = await self._get_guild(self.guild)
|
|
|
|
|
if isinstance(self.channel, int):
|
|
|
|
|
if not self.guild:
|
|
|
|
|
raise ValueError(f"[dpy_logger] cannot resolve channel {self.channel} without a guild")
|
|
|
|
|
self.channel = await self._get_channel(self.guild)
|
|
|
|
|
if self.channel is not None and not isinstance(self.channel, (discord.TextChannel, int)):
|
|
|
|
|
raise ValueError(f"[dpy_logger] channel {self.channel!r} is not a text channel")
|
|
|
|
|
if self.guild is None and isinstance(self.channel, discord.TextChannel):
|
|
|
|
|
self.guild = self.channel.guild
|
|
|
|
|
if self.guild is None and self.channel is not None:
|
|
|
|
|
raise ValueError("[dpy_logger] channel provided without a resolvable guild")
|
|
|
|
|
|
|
|
|
|
async def _get_guild(self, guild):
|
|
|
|
|
"""resolve a guild from id-or-object, raising if unresolvable"""
|
|
|
|
|
if guild:
|
|
|
|
|
if isinstance(guild, int):
|
|
|
|
|
resolved = self.bot.get_guild(guild)
|
|
|
|
|
if resolved is not None:
|
|
|
|
|
return resolved
|
|
|
|
|
# fetch_guild raises rather than returning None; normalize to ValueError
|
|
|
|
|
try:
|
|
|
|
|
return await self.bot.fetch_guild(guild)
|
|
|
|
|
except discord.HTTPException as error:
|
|
|
|
|
raise ValueError(f"[dpy_logger] failed to fetch guild {guild}: {error}") from error
|
|
|
|
|
if isinstance(guild, discord.Guild):
|
|
|
|
|
return guild
|
|
|
|
|
raise ValueError("[dpy_logger] no guild available for logging")
|
|
|
|
|
|
|
|
|
|
async def _get_channel(self, guild, override=False):
|
|
|
|
|
"""resolve the log channel, preferring self.channel when the guild matches"""
|
|
|
|
|
if not guild:
|
|
|
|
|
raise ValueError("[dpy_logger] cannot resolve channel without a guild")
|
|
|
|
|
|
|
|
|
|
if not override and getattr(guild, "id", guild) == getattr(self.guild, "id", self.guild):
|
|
|
|
|
if isinstance(self.channel, discord.TextChannel):
|
|
|
|
|
return self.channel
|
|
|
|
|
if isinstance(self.channel, int):
|
|
|
|
|
# fetch_channel raises NotFound/Forbidden/HTTPException/InvalidData (a
|
|
|
|
|
# ClientException, not an HTTPException); normalize all to ValueError
|
|
|
|
|
try:
|
|
|
|
|
channel = await guild.fetch_channel(self.channel)
|
|
|
|
|
except (discord.HTTPException, discord.ClientException) as error:
|
|
|
|
|
raise ValueError(f"[dpy_logger] could not fetch channel {self.channel}: {error}") from error
|
|
|
|
|
if not isinstance(channel, discord.TextChannel):
|
|
|
|
|
raise ValueError(f"[dpy_logger] channel {self.channel} is not a text channel")
|
|
|
|
|
return channel
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
channel_id = self.bot.settings[guild.id]["channels"]["logs"]
|
|
|
|
|
except KeyError as error:
|
|
|
|
|
raise ValueError(f"[dpy_logger] no log channel configured for guild {guild.id}") from error
|
|
|
|
|
try:
|
|
|
|
|
channel = await guild.fetch_channel(channel_id)
|
|
|
|
|
except (discord.HTTPException, discord.ClientException) as error:
|
|
|
|
|
raise ValueError(f"[dpy_logger] could not fetch channel {channel_id}: {error}") from error
|
|
|
|
|
if not isinstance(channel, discord.TextChannel):
|
|
|
|
|
raise ValueError(f"[dpy_logger] configured channel {channel_id} is not a text channel")
|
|
|
|
|
return channel
|
|
|
|
|
|
|
|
|
|
@staticmethod
|
|
|
|
|
def _cap_field(value: str, reserve: int = 0) -> str:
|
|
|
|
|
"""cap a value to discord's 1024-char field limit (less `reserve` chars for any wrapping
|
|
|
|
|
the caller adds around it, e.g. backticks), truncating with an ellipsis"""
|
|
|
|
|
limit = DISCORD_FIELD_VALUE_LIMIT - reserve
|
|
|
|
|
if len(value) > limit:
|
|
|
|
|
return value[:limit - 3] + "..."
|
|
|
|
|
return value
|
|
|
|
|
|
|
|
|
|
def build_embed(self, level, action, actor, details):
|
|
|
|
|
"""build the embed for a log call
|
|
|
|
|
|
|
|
|
|
if an `embed_builder` callable was passed at construction, it is used:
|
|
|
|
|
embed_builder(logger, level, action, actor, details) -> discord.Embed
|
|
|
|
|
otherwise the default structure is built. subclasses may also override
|
|
|
|
|
this method directly instead of passing a callable.
|
|
|
|
|
"""
|
|
|
|
|
if self._embed_builder is not None:
|
|
|
|
|
return self._embed_builder(self, level, action, actor, details)
|
|
|
|
|
em = discord.Embed(color=self.colors[level])
|
|
|
|
|
# discord 400s an empty or >1024-char field value; cap then wrap so the closing
|
|
|
|
|
# backtick is never truncated off (cap the inner value, accounting for the 2 backticks)
|
|
|
|
|
if action is not None and str(action) != "":
|
|
|
|
|
em.add_field(name="Action", value=f"`{self._cap_field(str(action), reserve=2)}`", inline=True)
|
|
|
|
|
if actor is not None and str(actor) != "":
|
|
|
|
|
em.add_field(name="Actor", value=f"`{self._cap_field(str(actor), reserve=2)}`", inline=True)
|
|
|
|
|
log_value = str(details) if details is not None and str(details) != "" else "(no message)"
|
|
|
|
|
em.add_field(name="Log", value=self._cap_field(log_value), inline=False)
|
|
|
|
|
em.timestamp = datetime.now(self.timezone)
|
|
|
|
|
em.set_footer(text=f"{self.footer} Logging".strip(), icon_url=self.avatar)
|
|
|
|
|
return em
|
|
|
|
|
|
|
|
|
|
async def _resolve(self, guild):
|
|
|
|
|
"""resolve the target channel for a call, honoring a per-call guild override"""
|
|
|
|
|
return await self._get_channel(await self._get_guild(guild or self.guild), bool(guild))
|
|
|
|
|
|
|
|
|
|
def _emit_stdlib(self, level, action, actor, log_msg):
|
|
|
|
|
"""mirror the log line to the stdlib logger (routed to file/console by the app)"""
|
|
|
|
|
raw = (action, str(actor) if actor is not None else None, log_msg)
|
|
|
|
|
parts = [str(p) for p in raw if p is not None and str(p) != ""]
|
|
|
|
|
_log.log(LEVEL_MAP.get(level, logging.INFO), f"[{level}] " + " | ".join(parts))
|
|
|
|
|
|
|
|
|
|
async def _send(self, level, log_msg, action=None, actor=None, guild=None,
|
|
|
|
|
log_to_file=None, content=None):
|
|
|
|
|
"""resolve channel and dispatch a leveled embed; mirror to stdlib unless opted out"""
|
|
|
|
|
if self.log_to_file if log_to_file is None else log_to_file:
|
|
|
|
|
self._emit_stdlib(level, action, actor, log_msg)
|
|
|
|
|
try:
|
|
|
|
|
channel = await self._resolve(guild)
|
|
|
|
|
return await channel.send(
|
|
|
|
|
content=content,
|
|
|
|
|
embed=self.build_embed(level, action, actor, log_msg),
|
|
|
|
|
)
|
|
|
|
|
except Exception:
|
|
|
|
|
_log.exception(f"[dpy_logger] failed to send {level} log: {log_msg}")
|
|
|
|
|
return None
|
|
|
|
|
|
|
|
|
|
async def debug(self, log, action=None, actor=None, guild=None, log_to_file=None):
|
|
|
|
|
"""log a debug-level message"""
|
|
|
|
|
return await self._send("debug", log, action, actor, guild, log_to_file)
|
|
|
|
|
|
|
|
|
|
async def info(self, log, action=None, actor=None, guild=None, log_to_file=None):
|
|
|
|
|
"""log an info-level message"""
|
|
|
|
|
return await self._send("info", log, action, actor, guild, log_to_file)
|
|
|
|
|
|
|
|
|
|
async def success(self, log, action=None, actor=None, guild=None, log_to_file=None):
|
|
|
|
|
"""log a success-level message"""
|
|
|
|
|
return await self._send("success", log, action, actor, guild, log_to_file)
|
|
|
|
|
|
|
|
|
|
async def fail(self, log, action=None, actor=None, guild=None, log_to_file=None):
|
|
|
|
|
"""log a fail-level message"""
|
|
|
|
|
return await self._send("fail", log, action, actor, guild, log_to_file)
|
|
|
|
|
|
|
|
|
|
failure = fail
|
|
|
|
|
|
|
|
|
|
async def task(self, log, action=None, actor=None, guild=None, log_to_file=None):
|
|
|
|
|
"""log a task-level message with SYSTEM/TASK as the actor
|
|
|
|
|
|
|
|
|
|
actor is accepted for caller compatibility and ignored - task actions are
|
|
|
|
|
always attributed to SYSTEM/TASK regardless of the caller-supplied actor
|
|
|
|
|
"""
|
|
|
|
|
return await self._send("task", log, action, "SYSTEM/TASK", guild, log_to_file)
|
|
|
|
|
|
|
|
|
|
async def critical(self, log, action=None, actor=None, guild=None, log_to_file=None):
|
|
|
|
|
"""log a critical-level message and ping configured devs
|
|
|
|
|
|
|
|
|
|
delegates to _send with the ping/@here content so the send + error contract has a
|
|
|
|
|
single source of truth; critical only adds the alert `content`.
|
|
|
|
|
"""
|
|
|
|
|
if self.pings:
|
|
|
|
|
content = "alert: " + " ".join(f"<@{u}>" for u in self.pings)
|
|
|
|
|
elif self.alert_here:
|
|
|
|
|
content = "alert: @here"
|
|
|
|
|
else:
|
|
|
|
|
content = None
|
|
|
|
|
if content is not None and len(content) > DISCORD_CONTENT_LIMIT:
|
|
|
|
|
content = content[:DISCORD_CONTENT_LIMIT - 3] + "..."
|
|
|
|
|
return await self._send("critical", log, action, actor, guild, log_to_file, content)
|