1 Commits
Author SHA1 Message Date
dsqlandClaude Opus 4.8 9b6cf3bd05 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.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: disqualifier <dev@disqualifier.me>
2026-06-24 21:24:39 -04:00
5 changed files with 80 additions and 131 deletions
+1 -1
View File
@@ -1,5 +1,5 @@
# claude
.claude/
CLAUDE.md
# python
__pycache__/
+9 -20
View File
@@ -10,19 +10,17 @@ live from `bot.settings` so it can change at runtime via a command.
`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:
```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).
Drop the `@v1.0.0` suffix from the line above to install the latest unpinned.
## Usage
```python
@@ -72,27 +70,18 @@ await bot.log.debug("noisy", log_to_file=False) # -> Discord only
## Errors
Resolution failures (unresolvable guild/channel, bad config, a non-text channel — whether
passed as an id or as an already-resolved object — or a `timezone` that isn't a `tzinfo`
instance) raise `ValueError` from `initialize()` — a misconfigured logger should fail
loudly at setup. Underlying `discord` exceptions
(`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.
Resolution failures (unresolvable guild/channel, bad config) raise `ValueError` at
`initialize`/call time — a misconfigured logger should fail loudly at setup. 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.
## Construction contract
The host injects everything; the lib never imports `config`:
- `colors` (dict, optional) — per-level colors, merged over defaults
- `pings` (list of user ids) — mentioned on `critical()`; the ping content is capped at
Discord's 2000-char message limit so a long list still sends
- `timezone` (`tzinfo`, optional) — must be a real `tzinfo` instance (e.g. `datetime.timezone.utc`);
validated at `initialize()`, not a string
- `footer`, `avatar` — embed identity
- `pings` (list of user ids) — mentioned on `critical()`
- `timezone`, `footer`, `avatar` — embed identity
- `alert_here` (bool) — if no `pings` are set, `critical()` falls back to
`@here` only when this is `True`; otherwise it sends no mention
- `embed_builder` (callable, optional) — restyle embeds without subclassing
@@ -147,4 +136,4 @@ class FeedLogger(DPYLogger):
## 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]
name = "dpy_logger"
version = "1.1.0"
version = "0.1.0"
description = "Leveled Discord channel logger for discord.py — config-free, injectable, installable."
requires-python = ">=3.10"
dependencies = [
+1 -8
View File
@@ -1,10 +1,3 @@
from importlib.metadata import version, PackageNotFoundError
from .dpy_logger import DPYLogger, DEFAULT_COLORS
try:
__version__ = version("dpy_logger")
except PackageNotFoundError:
__version__ = "0.0.0+unknown"
__all__ = ["DPYLogger", "DEFAULT_COLORS", "__version__"]
__all__ = ["DPYLogger", "DEFAULT_COLORS"]
+68 -101
View File
@@ -1,10 +1,12 @@
"""
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).
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
@@ -19,27 +21,34 @@ read live from bot.settings at call time (no restart needed).
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.
levels: debug, info, success, fail (alias failure), task, critical.
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.
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.
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.
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, tzinfo
from datetime import datetime
from typing import Callable, Optional
_log = logging.getLogger(__name__)
@@ -53,9 +62,6 @@ DEFAULT_COLORS = {
"critical": 0x000000,
}
DISCORD_CONTENT_LIMIT = 2000
DISCORD_FIELD_VALUE_LIMIT = 1024
LEVEL_MAP = {
"debug": logging.DEBUG,
"info": logging.INFO,
@@ -98,33 +104,21 @@ class DPYLogger:
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
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")
@@ -134,40 +128,20 @@ class DPYLogger:
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 not override and guild == 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
return await guild.fetch_channel(self.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
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
@@ -180,14 +154,11 @@ class DPYLogger:
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)
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
@@ -198,26 +169,22 @@ class DPYLogger:
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) != ""]
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, content=None):
"""resolve channel and dispatch a leveled embed; mirror to stdlib unless opted out"""
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(
content=content,
embed=self.build_embed(level, action, actor, 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 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):
@@ -238,26 +205,26 @@ class DPYLogger:
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
"""
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
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`.
"""
"""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
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)
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