add package: pyproject + src
DPYLogger: leveled discord channel logger (debug/info/success/fail/task/ critical) over discord.py. config-free — embed identity injected at construction, per-guild channel routing read live from bot.settings. dual-sink: mirrors every call to the stdlib logger. embed_builder callable + build_embed override for customization. raises by design; object-only (no module proxy). src/ layout, hatchling build. Signed-off-by: disqualifier <dev@disqualifier.me>
This commit is contained in:
parent
a90a0061b0
commit
7a5684e065
15
pyproject.toml
Normal file
15
pyproject.toml
Normal file
@ -0,0 +1,15 @@
|
||||
[build-system]
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[project]
|
||||
name = "dpy_logger"
|
||||
version = "0.1.0"
|
||||
description = "Leveled Discord channel logger for discord.py — config-free, injectable, installable."
|
||||
requires-python = ">=3.10"
|
||||
dependencies = [
|
||||
"discord.py>=2.0",
|
||||
]
|
||||
|
||||
[tool.hatch.build.targets.wheel]
|
||||
packages = ["src/dpy_logger"]
|
||||
3
src/dpy_logger/__init__.py
Normal file
3
src/dpy_logger/__init__.py
Normal file
@ -0,0 +1,3 @@
|
||||
from .dpy_logger import DPYLogger, DEFAULT_COLORS
|
||||
|
||||
__all__ = ["DPYLogger", "DEFAULT_COLORS"]
|
||||
230
src/dpy_logger/dpy_logger.py
Normal file
230
src/dpy_logger/dpy_logger.py
Normal file
@ -0,0 +1,230 @@
|
||||
"""
|
||||
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: all static identity is injected at construction; dynamic
|
||||
per-guild channel routing is read live from bot.settings at call time, so a
|
||||
command that mutates bot.settings changes routing without a restart.
|
||||
|
||||
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.
|
||||
|
||||
dynamic routing: a per-call guild= argument logs to that guild's configured
|
||||
channel via bot.settings[guild.id]['channels']['logs']; omit it to use the
|
||||
channel this logger was constructed with.
|
||||
|
||||
custom embeds: pass embed_builder=fn to restyle without subclassing, where
|
||||
fn(logger, level, action, actor, details) -> discord.Embed. every level
|
||||
(including critical) routes through it. for complex cases override the
|
||||
build_embed method in a subclass instead.
|
||||
|
||||
dual sink: every call also mirrors to the stdlib logger (getLogger(__name__)),
|
||||
which the app routes to file/console. set log_to_file=False at construction to
|
||||
disable, or pass log_to_file=True/False per call to override. the stdlib emit
|
||||
happens before the discord send, so the record survives even if discord fails.
|
||||
|
||||
extending: this base has no feed/announcement method by design. a project
|
||||
that wants one subclasses DPYLogger and adds it, reusing _resolve.
|
||||
|
||||
errors: resolution failures (unresolvable guild/channel, bad config) raise
|
||||
ValueError at initialize/call time — a misconfigured logger should fail loudly.
|
||||
per-call send failures do NOT 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
|
||||
from typing import Callable, Optional
|
||||
|
||||
_log = logging.getLogger(__name__)
|
||||
|
||||
DEFAULT_COLORS = {
|
||||
"debug": 0x808080,
|
||||
"info": 0x709288,
|
||||
"success": 0x00E27D,
|
||||
"fail": 0xD93025,
|
||||
"task": 0x5AC6E1,
|
||||
"critical": 0x000000,
|
||||
}
|
||||
|
||||
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 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)
|
||||
|
||||
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) or await self.bot.fetch_guild(guild)
|
||||
if not resolved:
|
||||
raise ValueError(f"[dpy_logger] failed to fetch guild {guild}")
|
||||
return resolved
|
||||
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 guild == self.guild:
|
||||
if isinstance(self.channel, discord.TextChannel):
|
||||
return self.channel
|
||||
if isinstance(self.channel, int):
|
||||
return await guild.fetch_channel(self.channel)
|
||||
|
||||
try:
|
||||
channel_id = self.bot.settings[guild.id]["channels"]["logs"]
|
||||
channel = await guild.fetch_channel(channel_id)
|
||||
if not isinstance(channel, discord.TextChannel):
|
||||
raise ValueError(f"[dpy_logger] configured channel {channel_id} is not a text channel")
|
||||
return channel
|
||||
except KeyError:
|
||||
raise ValueError(f"[dpy_logger] no log channel configured for guild {guild.id}")
|
||||
|
||||
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])
|
||||
if action:
|
||||
em.add_field(name="Action", value=f"`{action}`", inline=True)
|
||||
if actor:
|
||||
em.add_field(name="Actor", value=f"`{actor}`", inline=True)
|
||||
em.add_field(name="Log", value=details, 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)"""
|
||||
parts = [p for p in (action, str(actor) if actor else None, log_msg) if 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):
|
||||
"""resolve channel and dispatch a leveled embed; mirror to stdlib unless opted out
|
||||
|
||||
stdlib emit happens first so the record survives even if the discord send fails.
|
||||
a failed send falls back to stdlib rather than propagating into the caller.
|
||||
"""
|
||||
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(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, guild=None, log_to_file=None):
|
||||
"""log a task-level message with SYSTEM/TASK as the 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"""
|
||||
if self.log_to_file if log_to_file is None else log_to_file:
|
||||
self._emit_stdlib("critical", action, actor, log)
|
||||
if self.pings:
|
||||
content = "alert: " + " ".join(f"<@{u}>" for u in self.pings)
|
||||
elif self.alert_here:
|
||||
content = "alert: @here"
|
||||
else:
|
||||
content = None
|
||||
try:
|
||||
channel = await self._resolve(guild)
|
||||
return await channel.send(
|
||||
content=content,
|
||||
embed=self.build_embed("critical", action, actor, log),
|
||||
)
|
||||
except Exception:
|
||||
_log.exception(f"[dpy_logger] failed to send critical log: {log}")
|
||||
return None
|
||||
Loading…
Reference in New Issue
Block a user