2 Commits
Author SHA1 Message Date
dsqlandClaude Opus 4.8 7935ea4511 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.
embed_builder callable + build_embed override for customization. raises
by design; object-only (no module proxy). src/ layout, hatchling build.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: disqualifier <dev@disqualifier.me>
2026-06-23 23:08:35 -04:00
dsql 48de7d7065 init: leveled discord logger
Signed-off-by: disqualifier <dev@disqualifier.me>
2026-06-23 15:12:03 -04:00
5 changed files with 90 additions and 141 deletions
+1 -1
View File
@@ -1,5 +1,5 @@
# claude # claude
.claude/ CLAUDE.md
# python # python
__pycache__/ __pycache__/
+14 -25
View File
@@ -10,19 +10,17 @@ live from `bot.settings` so it can change at runtime via a command.
`requirements.txt`: `requirements.txt`:
``` ```
dpy_logger @ git+ssh://git@git.rethinkstudios.io/rethink-public/dpy_logger.git@v1.0.0 dpy_logger @ git+ssh://git@git.rethinkstudios.io/rethink-public/dpy_logger.git@v0.1.0
``` ```
Direct: Direct:
```bash ```bash
pip install "dpy_logger @ git+ssh://git@git.rethinkstudios.io/rethink-public/dpy_logger.git@v1.0.0" pip install "dpy_logger @ git+ssh://git@git.rethinkstudios.io/rethink-public/dpy_logger.git@v0.1.0"
``` ```
Requires `discord.py` (pulled transitively). Requires `discord.py` (pulled transitively).
Drop the `@v1.0.0` suffix from the line above to install the latest unpinned.
## Usage ## Usage
```python ```python
@@ -30,11 +28,11 @@ from dpy_logger import DPYLogger
bot.log = DPYLogger( bot.log = DPYLogger(
bot, guild_id, channel_id, bot, guild_id, channel_id,
colors=log_colors, # optional; merged over sensible defaults colors=cfg.log_colors, # optional; merged over sensible defaults
pings=authorized_devs, # mentioned on critical() pings=cfg.authorized_devs, # mentioned on critical()
timezone=tz, timezone=cfg.timezone,
footer=bot_footer, footer=cfg.bot_footer,
avatar=bot_avatar, avatar=cfg.bot_avatar,
) )
await bot.log.initialize() # resolves ids -> objects, call once await bot.log.initialize() # resolves ids -> objects, call once
@@ -72,27 +70,18 @@ await bot.log.debug("noisy", log_to_file=False) # -> Discord only
## Errors ## Errors
Resolution failures (unresolvable guild/channel, bad config, a non-text channel — whether Resolution failures (unresolvable guild/channel, bad config) raise `ValueError` at
passed as an id or as an already-resolved object — or a `timezone` that isn't a `tzinfo` `initialize`/call time — a misconfigured logger should fail loudly at setup. Per-call
instance) raise `ValueError` from `initialize()` — a misconfigured logger should fail **send** failures do **not** propagate: they fall back to the stdlib logger so a
loudly at setup. Underlying `discord` exceptions transient Discord failure never breaks the caller's command.
(`NotFound` / `Forbidden` / `HTTPException` / `InvalidData` and other `ClientException`
subclasses) from `fetch_guild`/`fetch_channel` are normalized to that `ValueError` on every
resolution path (construction channel, per-guild settings lookup) so callers see one error
type. On a **per-call** send, neither resolution nor send failures propagate: they fall
back to the stdlib logger so a transient Discord failure (or a per-call `guild=` that
doesn't resolve) never breaks the caller's command.
## Construction contract ## Construction contract
The host injects everything; the lib never imports `config`: The host injects everything; the lib never imports `config`:
- `colors` (dict, optional) — per-level colors, merged over defaults - `colors` (dict, optional) — per-level colors, merged over defaults
- `pings` (list of user ids) — mentioned on `critical()`; the ping content is capped at - `pings` (list of user ids) — mentioned on `critical()`
Discord's 2000-char message limit so a long list still sends - `timezone`, `footer`, `avatar` — embed identity
- `timezone` (`tzinfo`, optional) — must be a real `tzinfo` instance (e.g. `datetime.timezone.utc`);
validated at `initialize()`, not a string
- `footer`, `avatar` — embed identity
- `alert_here` (bool) — if no `pings` are set, `critical()` falls back to - `alert_here` (bool) — if no `pings` are set, `critical()` falls back to
`@here` only when this is `True`; otherwise it sends no mention `@here` only when this is `True`; otherwise it sends no mention
- `embed_builder` (callable, optional) — restyle embeds without subclassing - `embed_builder` (callable, optional) — restyle embeds without subclassing
@@ -147,4 +136,4 @@ class FeedLogger(DPYLogger):
## Versioning ## Versioning
Releases are tagged `vX.Y.Z`. The install line above pins a release; drop the `@vX.Y.Z` suffix to install the latest unpinned. Pin deliberately for reproducible installs. Tagged `vX.Y.Z`. Pin the tag in `requirements.txt`.
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project] [project]
name = "dpy_logger" name = "dpy_logger"
version = "1.1.0" version = "0.1.0"
description = "Leveled Discord channel logger for discord.py — config-free, injectable, installable." description = "Leveled Discord channel logger for discord.py — config-free, injectable, installable."
requires-python = ">=3.10" requires-python = ">=3.10"
dependencies = [ dependencies = [
+1 -8
View File
@@ -1,10 +1,3 @@
from importlib.metadata import version, PackageNotFoundError
from .dpy_logger import DPYLogger, DEFAULT_COLORS from .dpy_logger import DPYLogger, DEFAULT_COLORS
try: __all__ = ["DPYLogger", "DEFAULT_COLORS"]
__version__ = version("dpy_logger")
except PackageNotFoundError:
__version__ = "0.0.0+unknown"
__all__ = ["DPYLogger", "DEFAULT_COLORS", "__version__"]
+69 -102
View File
@@ -1,45 +1,54 @@
""" """
discord channel logger discord channel logger
leveled logging to a discord channel via embeds. attach to your bot (e.g. leveled logging to a discord channel via embeds. attach to your bot
bot.log) and call bot.log.info(...), bot.log.critical(...), etc. config-free: (e.g. bot.log) and call bot.log.info(...), bot.log.critical(...), etc.
static identity is injected at construction; per-guild channel routing is
read live from bot.settings at call time (no restart needed). 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 from dpy_logger import DPYLogger
bot.log = DPYLogger( bot.log = DPYLogger(
bot, guild_id, channel_id, bot, guild_id, channel_id,
colors=log_colors, # optional, falls back to defaults colors=cfg.log_colors, # optional, falls back to defaults
pings=authorized_devs, # mentioned on critical() pings=cfg.authorized_devs, # mentioned on critical()
timezone=tz, timezone=cfg.timezone,
footer=bot_footer, footer=cfg.bot_footer,
avatar=bot_avatar, avatar=cfg.bot_avatar,
) )
await bot.log.initialize() # resolves ids -> objects await bot.log.initialize() # resolves ids -> objects
await bot.log.success("user promoted", action="promote", actor=ctx.author) await bot.log.success("user promoted", action="promote", actor=ctx.author)
levels: debug, info, success, fail (alias failure), task, critical. a per-call levels: debug, info, success, fail (alias failure), task, critical.
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__)) dynamic routing: a per-call guild= argument logs to that guild's configured
before the discord send, so the record survives even if discord fails. set channel via bot.settings[guild.id]['channels']['logs']; omit it to use the
log_to_file=False at construction, or per call, to disable. channel this logger was constructed with.
errors: setup raises, sends swallow. resolution failures (unresolvable custom embeds: pass embed_builder=fn to restyle without subclassing, where
guild/channel, bad config, a non-text channel, an invalid timezone) raise fn(logger, level, action, actor, details) -> discord.Embed. every level
ValueError from `initialize()`, with underlying discord exceptions normalized (including critical) routes through it. for complex cases override the
to it. per-call send failures never propagate - they fall back to the stdlib build_embed method in a subclass instead.
logger so a transient discord failure never breaks the caller's command.
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 logging
import discord import discord
from datetime import datetime, tzinfo from datetime import datetime
from typing import Callable, Optional from typing import Callable, Optional
_log = logging.getLogger(__name__) _log = logging.getLogger(__name__)
@@ -53,9 +62,6 @@ DEFAULT_COLORS = {
"critical": 0x000000, "critical": 0x000000,
} }
DISCORD_CONTENT_LIMIT = 2000
DISCORD_FIELD_VALUE_LIMIT = 1024
LEVEL_MAP = { LEVEL_MAP = {
"debug": logging.DEBUG, "debug": logging.DEBUG,
"info": logging.INFO, "info": logging.INFO,
@@ -98,33 +104,21 @@ class DPYLogger:
async def initialize(self): async def initialize(self):
"""resolve guild/channel from ids to objects; call once after construction""" """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): if isinstance(self.guild, int):
self.guild = await self._get_guild(self.guild) self.guild = await self._get_guild(self.guild)
if isinstance(self.channel, int): if isinstance(self.channel, int):
if not self.guild: if not self.guild:
raise ValueError(f"[dpy_logger] cannot resolve channel {self.channel} without a guild") raise ValueError(f"[dpy_logger] cannot resolve channel {self.channel} without a guild")
self.channel = await self._get_channel(self.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): async def _get_guild(self, guild):
"""resolve a guild from id-or-object, raising if unresolvable""" """resolve a guild from id-or-object, raising if unresolvable"""
if guild: if guild:
if isinstance(guild, int): if isinstance(guild, int):
resolved = self.bot.get_guild(guild) resolved = self.bot.get_guild(guild) or await self.bot.fetch_guild(guild)
if resolved is not None: if not resolved:
raise ValueError(f"[dpy_logger] failed to fetch guild {guild}")
return resolved 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): if isinstance(guild, discord.Guild):
return guild return guild
raise ValueError("[dpy_logger] no guild available for logging") raise ValueError("[dpy_logger] no guild available for logging")
@@ -134,40 +128,20 @@ class DPYLogger:
if not guild: if not guild:
raise ValueError("[dpy_logger] cannot resolve channel without a 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 not override and guild == self.guild:
if isinstance(self.channel, discord.TextChannel): if isinstance(self.channel, discord.TextChannel):
return self.channel return self.channel
if isinstance(self.channel, int): if isinstance(self.channel, int):
# fetch_channel raises NotFound/Forbidden/HTTPException/InvalidData (a return await guild.fetch_channel(self.channel)
# 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: try:
channel_id = self.bot.settings[guild.id]["channels"]["logs"] 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) 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): if not isinstance(channel, discord.TextChannel):
raise ValueError(f"[dpy_logger] configured channel {channel_id} is not a text channel") raise ValueError(f"[dpy_logger] configured channel {channel_id} is not a text channel")
return channel return channel
except KeyError:
@staticmethod raise ValueError(f"[dpy_logger] no log channel configured for guild {guild.id}")
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): def build_embed(self, level, action, actor, details):
"""build the embed for a log call """build the embed for a log call
@@ -180,14 +154,11 @@ class DPYLogger:
if self._embed_builder is not None: if self._embed_builder is not None:
return self._embed_builder(self, level, action, actor, details) return self._embed_builder(self, level, action, actor, details)
em = discord.Embed(color=self.colors[level]) em = discord.Embed(color=self.colors[level])
# discord 400s an empty or >1024-char field value; cap then wrap so the closing if action:
# backtick is never truncated off (cap the inner value, accounting for the 2 backticks) em.add_field(name="Action", value=f"`{action}`", inline=True)
if action is not None and str(action) != "": if actor:
em.add_field(name="Action", value=f"`{self._cap_field(str(action), reserve=2)}`", inline=True) em.add_field(name="Actor", value=f"`{actor}`", inline=True)
if actor is not None and str(actor) != "": em.add_field(name="Log", value=details, inline=False)
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.timestamp = datetime.now(self.timezone)
em.set_footer(text=f"{self.footer} Logging".strip(), icon_url=self.avatar) em.set_footer(text=f"{self.footer} Logging".strip(), icon_url=self.avatar)
return em return em
@@ -198,26 +169,22 @@ class DPYLogger:
def _emit_stdlib(self, level, action, actor, log_msg): def _emit_stdlib(self, level, action, actor, log_msg):
"""mirror the log line to the stdlib logger (routed to file/console by the app)""" """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 = [p for p in (action, str(actor) if actor else None, log_msg) if p]
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)) _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, async def _send(self, level, log_msg, action=None, actor=None, guild=None, log_to_file=None):
log_to_file=None, content=None): """resolve channel and dispatch a leveled embed; mirror to stdlib unless opted out
"""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: if self.log_to_file if log_to_file is None else log_to_file:
self._emit_stdlib(level, action, actor, log_msg) self._emit_stdlib(level, action, actor, log_msg)
try: try:
channel = await self._resolve(guild) channel = await self._resolve(guild)
return await channel.send( return await channel.send(embed=self.build_embed(level, action, actor, log_msg))
content=content, except Exception:
embed=self.build_embed(level, action, actor, log_msg), _log.exception(f"[dpy_logger] failed to send {level} log: {log_msg}")
)
except Exception as exc:
# WARNING, not ERROR: the record was already mirrored to the stdlib sink BEFORE the
# send (see _emit_stdlib above), so a swallowed discord-send failure is recovered
# cleanly - the log survives, only the channel mirror was lost. lazy interpolation.
_log.warning("[dpy_logger] failed to send %s log: %s (%s)", level, log_msg, exc)
return None return None
async def debug(self, log, action=None, actor=None, guild=None, log_to_file=None): async def debug(self, log, action=None, actor=None, guild=None, log_to_file=None):
@@ -238,26 +205,26 @@ class DPYLogger:
failure = fail failure = fail
async def task(self, log, action=None, actor=None, guild=None, log_to_file=None): async def task(self, log, action=None, guild=None, log_to_file=None):
"""log a task-level message with SYSTEM/TASK as the actor """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) 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): async def critical(self, log, action=None, actor=None, guild=None, log_to_file=None):
"""log a critical-level message and ping configured devs """log a critical-level message and ping configured devs"""
if self.log_to_file if log_to_file is None else log_to_file:
delegates to _send with the ping/@here content so the send + error contract has a self._emit_stdlib("critical", action, actor, log)
single source of truth; critical only adds the alert `content`.
"""
if self.pings: if self.pings:
content = "alert: " + " ".join(f"<@{u}>" for u in self.pings) content = "alert: " + " ".join(f"<@{u}>" for u in self.pings)
elif self.alert_here: elif self.alert_here:
content = "alert: @here" content = "alert: @here"
else: else:
content = None content = None
if content is not None and len(content) > DISCORD_CONTENT_LIMIT: try:
content = content[:DISCORD_CONTENT_LIMIT - 3] + "..." channel = await self._resolve(guild)
return await self._send("critical", log, action, actor, guild, log_to_file, content) 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