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
4 changed files with 65 additions and 116 deletions
+1 -1
View File
@@ -1,5 +1,5 @@
# claude # claude
.claude/ CLAUDE.md
# python # python
__pycache__/ __pycache__/
+9 -20
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@v0.1.6 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@v0.1.6" 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 `@v0.1.6` suffix from the line above to install the latest unpinned.
## Usage ## Usage
```python ```python
@@ -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 = "0.1.6" 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 = [
+54 -94
View File
@@ -1,10 +1,12 @@
""" """
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
@@ -23,32 +25,30 @@ levels: debug, info, success, fail (alias failure), task, critical.
dynamic routing: a per-call guild= argument logs to that guild's configured 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 via bot.settings[guild.id]['channels']['logs']; omit it to use the
construction channel. channel this logger was constructed with.
custom embeds: pass embed_builder=fn(logger, level, action, actor, details) custom embeds: pass embed_builder=fn to restyle without subclassing, where
-> discord.Embed to restyle without subclassing; every level routes through fn(logger, level, action, actor, details) -> discord.Embed. every level
it. for complex cases override build_embed in a subclass instead. (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__)); dual sink: every call also mirrors to the stdlib logger (getLogger(__name__)),
the stdlib emit happens before the discord send, so the record survives even which the app routes to file/console. set log_to_file=False at construction to
if discord fails. set log_to_file=False at construction, or per call, to disable, or pass log_to_file=True/False per call to override. the stdlib emit
disable. happens before the discord send, so the record survives even if discord fails.
extending: no feed/announcement method by design; subclass and reuse extending: this base has no feed/announcement method by design. a project
_resolve for project-specific log types. that wants one subclasses DPYLogger and adds it, reusing _resolve.
errors: setup raises, sends swallow. resolution failures (unresolvable errors: resolution failures (unresolvable guild/channel, bad config) raise
guild/channel, bad config, a non-text channel, an invalid timezone) raise ValueError at initialize/call time — a misconfigured logger should fail loudly.
ValueError from `initialize()`. underlying discord exceptions (NotFound/ per-call send failures do NOT propagate: they fall back to the stdlib logger so
Forbidden/HTTPException/InvalidData and other ClientException subclasses) a transient discord failure never breaks the caller's command.
are normalized to that same ValueError. on a per-call send, resolution and
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 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__)
@@ -62,8 +62,6 @@ DEFAULT_COLORS = {
"critical": 0x000000, "critical": 0x000000,
} }
DISCORD_CONTENT_LIMIT = 2000
LEVEL_MAP = { LEVEL_MAP = {
"debug": logging.DEBUG, "debug": logging.DEBUG,
"info": logging.INFO, "info": logging.INFO,
@@ -106,37 +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)
# a channel with no guild would otherwise blackout every send silently; derive
# the guild from it, or fail loud here rather than later
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")
# a wrong-type channel object (Thread/VoiceChannel/ForumChannel) would also pass
# silently and misroute/blackhole later; fail loud here instead
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")
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:
return resolved raise ValueError(f"[dpy_logger] failed to fetch guild {guild}")
# fetch_guild raises rather than returning None; normalize to ValueError return resolved
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")
@@ -146,31 +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"]
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: except KeyError:
raise ValueError(f"[dpy_logger] no log channel configured for guild {guild.id}") raise ValueError(f"[dpy_logger] no log channel configured for guild {guild.id}")
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
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
@@ -183,16 +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])
# is-not-None keeps falsy-but-valid values (0, False) rendering if action:
if action is not None and str(action) != "":
em.add_field(name="Action", value=f"`{action}`", inline=True) em.add_field(name="Action", value=f"`{action}`", inline=True)
if actor is not None and str(actor) != "": if actor:
em.add_field(name="Actor", value=f"`{actor}`", inline=True) em.add_field(name="Actor", value=f"`{actor}`", inline=True)
# discord 400s an empty or >1024-char field value; substitute + cap em.add_field(name="Log", value=details, inline=False)
log_value = str(details) if details is not None and str(details) != "" else "(no message)"
if len(log_value) > 1024:
log_value = log_value[:1021] + "..."
em.add_field(name="Log", value=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
@@ -203,26 +169,20 @@ 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 a failed discord send, which stdlib emit happens first so the record survives even if the discord send fails.
falls back to stdlib rather than propagating. `content` carries the critical-level a failed send falls back to stdlib rather than propagating into the caller.
ping/@here text (None for other levels), so every level shares this one contract.
""" """
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,
embed=self.build_embed(level, action, actor, log_msg),
)
except Exception: except Exception:
_log.exception(f"[dpy_logger] failed to send {level} log: {log_msg}") _log.exception(f"[dpy_logger] failed to send {level} log: {log_msg}")
return None return None
@@ -245,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