add package: pyproject + src (parse/fit/sanitize/chunk/table/timestamp/confirm/choose/safe_send)
Signed-off-by: disqualifier <dev@disqualifier.me>
This commit is contained in:
@@ -0,0 +1,15 @@
|
||||
[build-system]
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[project]
|
||||
name = "dpy_commons"
|
||||
version = "0.1.0"
|
||||
description = "Shared discord.py utilities — message/embed parsing, limit-fitting, link extraction, chunking, timestamps, await-prompts, limit-safe send. Config-free, installable."
|
||||
requires-python = ">=3.10"
|
||||
dependencies = [
|
||||
"discord.py>=2.2",
|
||||
]
|
||||
|
||||
[tool.hatch.build.targets.wheel]
|
||||
packages = ["src/dpy_commons"]
|
||||
@@ -0,0 +1,61 @@
|
||||
"""dpy_commons — shared discord.py utilities (the discord-side sibling of commons)
|
||||
|
||||
a module of functions grouped by concern: message/embed parsing, embed sanitizing +
|
||||
limit-fitting, link extraction, text chunking, timestamp helpers, interactive await-prompts,
|
||||
and a limit-safe send. see each submodule's docstring for the contract.
|
||||
"""
|
||||
from .embeds import fit_embed, sanitize_embed, split_embeds
|
||||
from .limits import (
|
||||
EMBED_AUTHOR,
|
||||
EMBED_DESC,
|
||||
EMBED_FIELD_NAME,
|
||||
EMBED_FIELD_VALUE,
|
||||
EMBED_FOOTER,
|
||||
EMBED_MAX_FIELDS,
|
||||
EMBED_TITLE,
|
||||
EMBED_TOTAL,
|
||||
MSG_LIMIT,
|
||||
MSG_MAX_EMBEDS,
|
||||
)
|
||||
from .parsing import extract_message_links, parse_message, sanitize_mentions
|
||||
from .prompts import choose, confirm
|
||||
from .send import safe_send
|
||||
from .text import (
|
||||
chunk_text,
|
||||
discord_timestamp,
|
||||
format_table,
|
||||
humanize_delta,
|
||||
truncate,
|
||||
wrap_bare_links,
|
||||
)
|
||||
|
||||
__version__ = "0.1.0"
|
||||
|
||||
__all__ = [
|
||||
"parse_message",
|
||||
"extract_message_links",
|
||||
"sanitize_mentions",
|
||||
"fit_embed",
|
||||
"sanitize_embed",
|
||||
"split_embeds",
|
||||
"chunk_text",
|
||||
"format_table",
|
||||
"discord_timestamp",
|
||||
"humanize_delta",
|
||||
"truncate",
|
||||
"wrap_bare_links",
|
||||
"confirm",
|
||||
"choose",
|
||||
"safe_send",
|
||||
"MSG_LIMIT",
|
||||
"EMBED_TITLE",
|
||||
"EMBED_DESC",
|
||||
"EMBED_FIELD_NAME",
|
||||
"EMBED_FIELD_VALUE",
|
||||
"EMBED_FOOTER",
|
||||
"EMBED_AUTHOR",
|
||||
"EMBED_TOTAL",
|
||||
"EMBED_MAX_FIELDS",
|
||||
"MSG_MAX_EMBEDS",
|
||||
"__version__",
|
||||
]
|
||||
@@ -0,0 +1,135 @@
|
||||
"""embed helpers: fit an embed within discord's hard limits, sanitize one to be safe-to-send,
|
||||
and split a list into sendable groups; none of these mutate their input"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
import discord
|
||||
|
||||
from .limits import (
|
||||
EMBED_AUTHOR,
|
||||
EMBED_DESC,
|
||||
EMBED_FIELD_NAME,
|
||||
EMBED_FIELD_VALUE,
|
||||
EMBED_FOOTER,
|
||||
EMBED_MAX_FIELDS,
|
||||
EMBED_TITLE,
|
||||
EMBED_TOTAL,
|
||||
MSG_MAX_EMBEDS,
|
||||
)
|
||||
from .text import truncate, wrap_bare_links
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
_ZWSP = ""
|
||||
|
||||
|
||||
def _copy(embed: discord.Embed) -> discord.Embed:
|
||||
"""deep copy via dict round-trip so the input is never mutated"""
|
||||
return discord.Embed.from_dict(embed.to_dict())
|
||||
|
||||
|
||||
def fit_embed(embed: discord.Embed) -> discord.Embed:
|
||||
"""return a copy guaranteed within discord's hard limits: title/description/field/footer/
|
||||
author truncated to their caps, fields beyond 25 dropped, and the description then fields
|
||||
shrunk until the 6000 total fits; appends an ellipsis where it cuts, never mutates input"""
|
||||
out = _copy(embed)
|
||||
|
||||
if out.title:
|
||||
out.title = truncate(out.title, EMBED_TITLE)
|
||||
if out.description:
|
||||
out.description = truncate(out.description, EMBED_DESC)
|
||||
|
||||
footer = out.footer
|
||||
if footer and footer.text:
|
||||
out.set_footer(text=truncate(footer.text, EMBED_FOOTER), icon_url=footer.icon_url)
|
||||
author = out.author
|
||||
if author and author.name:
|
||||
out.set_author(name=truncate(author.name, EMBED_AUTHOR), url=author.url, icon_url=author.icon_url)
|
||||
|
||||
fields = list(out.fields)
|
||||
out.clear_fields()
|
||||
for field in fields[:EMBED_MAX_FIELDS]:
|
||||
out.add_field(
|
||||
name=truncate(field.name or _ZWSP, EMBED_FIELD_NAME),
|
||||
value=truncate(field.value or _ZWSP, EMBED_FIELD_VALUE),
|
||||
inline=field.inline,
|
||||
)
|
||||
|
||||
_shrink_to_total(out)
|
||||
return out
|
||||
|
||||
|
||||
def _shrink_to_total(embed: discord.Embed) -> None:
|
||||
"""truncate the description, then trailing fields, until len(embed) <= EMBED_TOTAL"""
|
||||
if len(embed) <= EMBED_TOTAL:
|
||||
return
|
||||
|
||||
if embed.description:
|
||||
overflow = len(embed) - EMBED_TOTAL
|
||||
keep = max(0, len(embed.description) - overflow)
|
||||
embed.description = truncate(embed.description, keep) if keep else _ZWSP
|
||||
if len(embed) <= EMBED_TOTAL:
|
||||
return
|
||||
|
||||
while len(embed) > EMBED_TOTAL and embed.fields:
|
||||
embed.remove_field(len(embed.fields) - 1)
|
||||
|
||||
|
||||
def sanitize_embed(embed: discord.Embed) -> discord.Embed:
|
||||
"""return a copy safe to send: non-discord links wrapped in <>, color normalized, empty
|
||||
strings discord rejects filled with a zero-width space, all other parts preserved, then
|
||||
run through fit_embed so the result is both clean and within limits; never mutates input"""
|
||||
out = _copy(embed)
|
||||
|
||||
if out.title:
|
||||
out.title = wrap_bare_links(out.title)
|
||||
out.description = wrap_bare_links(out.description) if out.description else out.description
|
||||
out.colour = _normalize_colour(out.colour)
|
||||
|
||||
footer = out.footer
|
||||
if footer and footer.text:
|
||||
out.set_footer(text=wrap_bare_links(footer.text), icon_url=footer.icon_url)
|
||||
author = out.author
|
||||
if author and author.name:
|
||||
out.set_author(name=wrap_bare_links(author.name), url=author.url, icon_url=author.icon_url)
|
||||
|
||||
fields = list(out.fields)
|
||||
out.clear_fields()
|
||||
for field in fields:
|
||||
out.add_field(
|
||||
name=wrap_bare_links(field.name) if field.name else _ZWSP,
|
||||
value=wrap_bare_links(field.value) if field.value else _ZWSP,
|
||||
inline=field.inline,
|
||||
)
|
||||
|
||||
if not out.description and not out.fields:
|
||||
out.description = _ZWSP
|
||||
|
||||
return fit_embed(out)
|
||||
|
||||
|
||||
def _normalize_colour(value: object) -> "discord.Colour | None":
|
||||
"""coerce a color-ish value to a discord.Colour: Colour passthrough, int, str '#hex',
|
||||
or an object with a .value; falls back to default() on bad input rather than raising"""
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, discord.Colour):
|
||||
return value
|
||||
try:
|
||||
if isinstance(value, int):
|
||||
return discord.Colour(value)
|
||||
if isinstance(value, str):
|
||||
return discord.Colour(int(value.lstrip("#"), 16))
|
||||
inner = getattr(value, "value", None)
|
||||
if isinstance(inner, int):
|
||||
return discord.Colour(inner)
|
||||
except (ValueError, TypeError):
|
||||
log.warning("dpy_commons: could not normalize embed colour %r, using default", value)
|
||||
return discord.Colour.default()
|
||||
|
||||
|
||||
def split_embeds(embeds: "list[discord.Embed]") -> "list[list[discord.Embed]]":
|
||||
"""chunk a list of embeds into groups of <= 10 (MSG_MAX_EMBEDS) so each group sends in one
|
||||
message"""
|
||||
return [embeds[i:i + MSG_MAX_EMBEDS] for i in range(0, len(embeds), MSG_MAX_EMBEDS)]
|
||||
@@ -0,0 +1,13 @@
|
||||
"""discord hard limits — single source of truth; every module references these, nothing
|
||||
hardcodes a limit inline"""
|
||||
|
||||
MSG_LIMIT = 2000
|
||||
EMBED_TITLE = 256
|
||||
EMBED_DESC = 4096
|
||||
EMBED_FIELD_NAME = 256
|
||||
EMBED_FIELD_VALUE = 1024
|
||||
EMBED_FOOTER = 2048
|
||||
EMBED_AUTHOR = 256
|
||||
EMBED_TOTAL = 6000
|
||||
EMBED_MAX_FIELDS = 25
|
||||
MSG_MAX_EMBEDS = 10
|
||||
@@ -0,0 +1,115 @@
|
||||
"""parsing / extraction: turn a discord.Message into a structured payload, pull jump-links
|
||||
out of text, and neutralize raw mention tokens; all pure — no bot/global reached for"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import re
|
||||
from typing import Union
|
||||
|
||||
import discord
|
||||
|
||||
from .embeds import fit_embed
|
||||
from .text import wrap_bare_links
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
_MESSAGE_LINK_RE = re.compile(
|
||||
r"https?://(?:ptb\.|canary\.)?discord(?:app)?\.com/channels/(\d+)/(\d+)/(\d+)"
|
||||
)
|
||||
_RAW_MENTION_RE = re.compile(r"<@[!&]?\d+>")
|
||||
|
||||
|
||||
def sanitize_mentions(text: str) -> str:
|
||||
"""neutralize pings in raw text: @everyone/@here -> (at)everyone/(at)here, and raw
|
||||
<@id>/<@&id> mention tokens -> (at)mention; pure text, resolves no names"""
|
||||
if not text:
|
||||
return text
|
||||
text = text.replace("@everyone", "(at)everyone").replace("@here", "(at)here")
|
||||
return _RAW_MENTION_RE.sub("(at)mention", text)
|
||||
|
||||
|
||||
def extract_message_links(
|
||||
text_or_message: "Union[str, discord.Message]",
|
||||
) -> "list[tuple[int, int, int]]":
|
||||
"""pull discord message jump-links out of text or a Message's content, returning
|
||||
(guild_id, channel_id, message_id) tuples"""
|
||||
text = text_or_message.content if isinstance(text_or_message, discord.Message) else text_or_message
|
||||
return [(int(g), int(c), int(m)) for g, c, m in _MESSAGE_LINK_RE.findall(text or "")]
|
||||
|
||||
|
||||
async def parse_message(message: discord.Message, *, strict: bool = False) -> dict:
|
||||
"""extract a discord.Message into a structured payload; pure, reaches for no bot/global
|
||||
|
||||
async because attachments are read from discord's cdn (network i/o) — the spec wrote this
|
||||
sync, but reading attachment bytes cannot be synchronous, so it is a coroutine (await it).
|
||||
|
||||
the only tolerated swallow is per-attachment: a read failure warns and skips by default so
|
||||
one bad attachment doesn't kill the parse; pass strict=True to raise on any such failure
|
||||
instead"""
|
||||
return {
|
||||
"content": wrap_bare_links(sanitize_mentions(message.content or "")),
|
||||
"mentions": _parse_mentions(message),
|
||||
"embeds": [fit_embed(embed) for embed in message.embeds],
|
||||
"attachments": await _parse_attachments(message, strict=strict),
|
||||
"stickers": [sticker.name for sticker in message.stickers],
|
||||
"poll": _parse_poll(message),
|
||||
"components": _parse_components(message),
|
||||
"reference": message.reference.message_id if message.reference else None,
|
||||
}
|
||||
|
||||
|
||||
def _parse_mentions(message: discord.Message) -> dict:
|
||||
"""collect user/role mentions plus everyone/here flags"""
|
||||
content = message.content or ""
|
||||
return {
|
||||
"users": [user.id for user in message.mentions],
|
||||
"roles": [role.id for role in message.role_mentions],
|
||||
"everyone": bool(message.mention_everyone) or "@everyone" in content,
|
||||
"here": "@here" in content,
|
||||
}
|
||||
|
||||
|
||||
async def _parse_attachments(message: discord.Message, *, strict: bool) -> "list[discord.File]":
|
||||
"""read each attachment into a discord.File; a read failure warns+skips unless strict"""
|
||||
files: "list[discord.File]" = []
|
||||
for attachment in message.attachments:
|
||||
try:
|
||||
files.append(await attachment.to_file())
|
||||
except (discord.HTTPException, discord.NotFound, OSError) as exc:
|
||||
if strict:
|
||||
raise
|
||||
log.warning("dpy_commons: skipping attachment %s: %s", attachment.filename, exc)
|
||||
return files
|
||||
|
||||
|
||||
def _parse_poll(message: discord.Message) -> "dict | None":
|
||||
"""extract a poll's question + option texts, or None if the message has no poll"""
|
||||
poll = getattr(message, "poll", None)
|
||||
if poll is None:
|
||||
return None
|
||||
return {
|
||||
"question": _poll_text(poll.question),
|
||||
"options": [_poll_text(answer) for answer in poll.answers],
|
||||
}
|
||||
|
||||
|
||||
def _poll_text(value: object) -> str:
|
||||
"""poll question/answer text across discord.py shapes (str, PollMedia, PollAnswer)"""
|
||||
if isinstance(value, str):
|
||||
return value
|
||||
media = getattr(value, "media", None)
|
||||
if media is not None and getattr(media, "text", None) is not None:
|
||||
return media.text
|
||||
return str(value)
|
||||
|
||||
|
||||
def _parse_components(message: discord.Message) -> "list[dict]":
|
||||
"""flatten action-row buttons to {label, url} entries"""
|
||||
buttons: "list[dict]" = []
|
||||
for row in message.components:
|
||||
for child in getattr(row, "children", []):
|
||||
label = getattr(child, "label", None)
|
||||
url = getattr(child, "url", None)
|
||||
if label is not None or url is not None:
|
||||
buttons.append({"label": label, "url": url})
|
||||
return buttons
|
||||
@@ -0,0 +1,218 @@
|
||||
"""interactive await-prompts: throw a prompt anywhere, await it, get the chosen VALUE back
|
||||
right there — no on_interaction listener, no view subclass, no state plumbing
|
||||
|
||||
built on a discord.py ``ui.View`` that internally ``await view.wait()``s and resolves to the
|
||||
selected value. the caller never touches discord's interaction machinery: ``confirm`` returns
|
||||
a bool (or None on timeout), ``choose`` returns the mapped value (or None). both scope to a
|
||||
user, disable their components after resolve/timeout, and fail loud on send/edit errors while
|
||||
treating a timeout as a normal None return.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any, Optional, Union
|
||||
|
||||
import discord
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
_MAX_BUTTONS = 5
|
||||
_NOT_FOR_YOU = "this prompt isn't for you"
|
||||
|
||||
Destination = Union[discord.abc.Messageable, discord.Interaction]
|
||||
EmojiInput = Union[str, discord.Emoji, discord.PartialEmoji]
|
||||
|
||||
|
||||
def _coerce_emoji(value: "Optional[EmojiInput]") -> "Optional[Union[str, discord.PartialEmoji]]":
|
||||
"""accept a unicode emoji, a custom-emoji string '<:name:id>', or a discord Emoji/
|
||||
PartialEmoji, and return what a component emoji= field accepts"""
|
||||
if value is None:
|
||||
return None
|
||||
if isinstance(value, (discord.Emoji, discord.PartialEmoji)):
|
||||
return value
|
||||
if isinstance(value, str):
|
||||
if value.startswith("<") and value.endswith(">"):
|
||||
return discord.PartialEmoji.from_str(value)
|
||||
return value
|
||||
raise ValueError(f"unsupported emoji input: {type(value).__name__}")
|
||||
|
||||
|
||||
def _infer_user(destination: Destination, user: "Optional[discord.abc.User]") -> "Optional[discord.abc.User]":
|
||||
"""derive the invoker to scope the prompt to, from an explicit user, an Interaction, or a
|
||||
ctx-like destination; None means anyone may click"""
|
||||
if user is not None:
|
||||
return user
|
||||
if isinstance(destination, discord.Interaction):
|
||||
return destination.user
|
||||
author = getattr(destination, "author", None)
|
||||
return author
|
||||
|
||||
|
||||
class _ValueButton(discord.ui.Button):
|
||||
"""a button that resolves the parent prompt to its bound value on click"""
|
||||
|
||||
def __init__(self, view: "_PromptView", value: Any, *, label=None, emoji=None, style=discord.ButtonStyle.secondary):
|
||||
super().__init__(label=label, emoji=emoji, style=style)
|
||||
self._prompt = view
|
||||
self._value = value
|
||||
|
||||
async def callback(self, interaction: discord.Interaction) -> None:
|
||||
"""resolve the prompt if the clicker is allowed, else tell them it isn't theirs"""
|
||||
if not await self._prompt._allowed(interaction):
|
||||
return
|
||||
await interaction.response.defer()
|
||||
self._prompt._resolve(self._value)
|
||||
|
||||
|
||||
class _ValueSelect(discord.ui.Select):
|
||||
"""a select whose chosen option resolves the parent prompt to a mapped value"""
|
||||
|
||||
def __init__(self, view: "_PromptView", mapping: "dict[str, Any]", placeholder: str):
|
||||
super().__init__(placeholder=placeholder, min_values=1, max_values=1)
|
||||
self._prompt = view
|
||||
self._mapping = mapping
|
||||
|
||||
async def callback(self, interaction: discord.Interaction) -> None:
|
||||
"""resolve to the mapped value of the selected option"""
|
||||
if not await self._prompt._allowed(interaction):
|
||||
return
|
||||
await interaction.response.defer()
|
||||
self._prompt._resolve(self._mapping[self.values[0]])
|
||||
|
||||
|
||||
class _PromptView(discord.ui.View):
|
||||
"""view backing an await-prompt: captures the resolved value and stops on first valid
|
||||
interaction or timeout"""
|
||||
|
||||
def __init__(self, *, user: "Optional[discord.abc.User]", timeout: float):
|
||||
super().__init__(timeout=timeout)
|
||||
self._user = user
|
||||
self.result: Any = None
|
||||
|
||||
async def _allowed(self, interaction: discord.Interaction) -> bool:
|
||||
"""true when the clicker may resolve the prompt; otherwise send an ephemeral notice"""
|
||||
if self._user is not None and interaction.user.id != self._user.id:
|
||||
await interaction.response.send_message(_NOT_FOR_YOU, ephemeral=True)
|
||||
return False
|
||||
return True
|
||||
|
||||
def _resolve(self, value: Any) -> None:
|
||||
"""record the result and stop waiting"""
|
||||
self.result = value
|
||||
self.stop()
|
||||
|
||||
|
||||
async def _send_prompt(destination: Destination, content: str, view: discord.ui.View) -> discord.Message:
|
||||
"""send the prompt to a Messageable or via an Interaction, returning the sent message"""
|
||||
if isinstance(destination, discord.Interaction):
|
||||
if destination.response.is_done():
|
||||
return await destination.followup.send(content, view=view, wait=True)
|
||||
await destination.response.send_message(content, view=view)
|
||||
return await destination.original_response()
|
||||
return await destination.send(content, view=view)
|
||||
|
||||
|
||||
async def _finish(message: discord.Message, view: _PromptView, cleanup: bool) -> None:
|
||||
"""disable the components and edit the message, or delete it when cleanup is set; edit/
|
||||
delete failures on an already-gone message are not errors"""
|
||||
if cleanup:
|
||||
try:
|
||||
await message.delete()
|
||||
except discord.NotFound:
|
||||
pass
|
||||
return
|
||||
for child in view.children:
|
||||
child.disabled = True
|
||||
try:
|
||||
await message.edit(view=view)
|
||||
except discord.NotFound:
|
||||
pass
|
||||
|
||||
|
||||
async def confirm(
|
||||
destination: Destination,
|
||||
prompt: str,
|
||||
*,
|
||||
user: "Optional[discord.abc.User]" = None,
|
||||
timeout: float = 60.0,
|
||||
yes: "EmojiInput" = "✅",
|
||||
no: "EmojiInput" = "❌",
|
||||
cleanup: bool = False,
|
||||
) -> "Optional[bool]":
|
||||
"""send prompt with a yes and a no button, await the click, and return True / False /
|
||||
None(timeout); scopes to user, disables the buttons after, and fails loud on send/edit"""
|
||||
scoped = _infer_user(destination, user)
|
||||
view = _PromptView(user=scoped, timeout=timeout)
|
||||
view.add_item(_ValueButton(view, True, emoji=_coerce_emoji(yes), style=discord.ButtonStyle.success))
|
||||
view.add_item(_ValueButton(view, False, emoji=_coerce_emoji(no), style=discord.ButtonStyle.danger))
|
||||
|
||||
message = await _send_prompt(destination, prompt, view)
|
||||
await view.wait()
|
||||
await _finish(message, view, cleanup)
|
||||
return view.result
|
||||
|
||||
|
||||
async def choose(
|
||||
destination: Destination,
|
||||
prompt: str,
|
||||
options: "dict[Any, Any]",
|
||||
*,
|
||||
user: "Optional[discord.abc.User]" = None,
|
||||
timeout: float = 60.0,
|
||||
cleanup: bool = False,
|
||||
) -> "Optional[Any]":
|
||||
"""render a button per option (or a select dropdown for >5 options or long labels), await
|
||||
the click, and return the chosen option's VALUE or None(timeout); options maps an emoji
|
||||
(unicode / '<:name:id>' / Emoji) or a label to a return value"""
|
||||
if not options:
|
||||
raise ValueError("choose requires at least one option")
|
||||
|
||||
scoped = _infer_user(destination, user)
|
||||
view = _PromptView(user=scoped, timeout=timeout)
|
||||
keys = list(options.keys())
|
||||
use_select = len(keys) > _MAX_BUTTONS or any(_is_long_label(k) for k in keys)
|
||||
|
||||
if use_select:
|
||||
mapping: "dict[str, Any]" = {}
|
||||
select = _ValueSelect(view, mapping, placeholder=prompt[:100])
|
||||
for i, key in enumerate(keys):
|
||||
token = str(i)
|
||||
mapping[token] = options[key]
|
||||
label, emoji = _split_key(key)
|
||||
select.add_option(label=label, value=token, emoji=emoji)
|
||||
view.add_item(select)
|
||||
else:
|
||||
for key in keys:
|
||||
label, emoji = _split_key(key)
|
||||
view.add_item(_ValueButton(view, options[key], label=label if not emoji else None, emoji=emoji))
|
||||
|
||||
message = await _send_prompt(destination, prompt, view)
|
||||
await view.wait()
|
||||
await _finish(message, view, cleanup)
|
||||
return view.result
|
||||
|
||||
|
||||
def _is_long_label(key: Any) -> bool:
|
||||
"""true when a plain-text label is too long for a button (>80 chars)"""
|
||||
if isinstance(key, str) and not (key.startswith("<") and key.endswith(">")):
|
||||
return len(key) > 80
|
||||
return False
|
||||
|
||||
|
||||
def _split_key(key: Any) -> "tuple[Optional[str], Optional[Union[str, discord.PartialEmoji]]]":
|
||||
"""turn an options key into (label, emoji): an emoji-looking key becomes the emoji with no
|
||||
label, everything else is a text label"""
|
||||
if isinstance(key, (discord.Emoji, discord.PartialEmoji)):
|
||||
return None, key
|
||||
if isinstance(key, str):
|
||||
if key.startswith("<") and key.endswith(">"):
|
||||
return None, discord.PartialEmoji.from_str(key)
|
||||
if _looks_unicode_emoji(key):
|
||||
return None, key
|
||||
return key, None
|
||||
return str(key), None
|
||||
|
||||
|
||||
def _looks_unicode_emoji(value: str) -> bool:
|
||||
"""heuristic: a short non-ascii token is treated as a unicode emoji key"""
|
||||
return bool(value) and len(value) <= 4 and not value.isascii()
|
||||
@@ -0,0 +1,55 @@
|
||||
"""limit-safe send that composes the text + embed helpers: chunk the content across
|
||||
messages, fit every embed and split into <=10 groups, and send as many messages as needed"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Optional
|
||||
|
||||
import discord
|
||||
|
||||
from .embeds import fit_embed, split_embeds
|
||||
from .text import chunk_text
|
||||
|
||||
|
||||
async def safe_send(
|
||||
destination: discord.abc.Messageable,
|
||||
content: "Optional[str]" = None,
|
||||
*,
|
||||
embeds: "Optional[list[discord.Embed]]" = None,
|
||||
**kwargs,
|
||||
) -> "list[discord.Message]":
|
||||
"""send that never fails on a discord limit: chunk_text the content across messages,
|
||||
fit_embed every embed and split_embeds into <=10 groups, and send as many messages as
|
||||
needed; returns the sent Messages, fails loud on perms/HTTP errors (only limit-handling is
|
||||
automatic)
|
||||
|
||||
extra kwargs ride only on the FIRST message so per-send options (files, reference, etc.)
|
||||
aren't duplicated across the split"""
|
||||
content_chunks = chunk_text(content) if content else []
|
||||
embed_groups = split_embeds([fit_embed(embed) for embed in embeds]) if embeds else []
|
||||
|
||||
messages: "list[discord.Message]" = []
|
||||
first = True
|
||||
|
||||
for i, chunk in enumerate(content_chunks):
|
||||
# attach the first embed group to the last content message so a single content+embeds
|
||||
# call collapses to one message when it fits
|
||||
group = embed_groups.pop(0) if (i == len(content_chunks) - 1 and embed_groups) else None
|
||||
messages.append(await _send(destination, chunk, group, first, kwargs))
|
||||
first = False
|
||||
|
||||
for group in embed_groups:
|
||||
messages.append(await _send(destination, None, group, first, kwargs))
|
||||
first = False
|
||||
|
||||
if not messages:
|
||||
messages.append(await _send(destination, content or "", None, True, kwargs))
|
||||
|
||||
return messages
|
||||
|
||||
|
||||
async def _send(destination, content, embeds, first, kwargs) -> discord.Message:
|
||||
"""send one message; first-message kwargs are applied once then dropped"""
|
||||
extra = dict(kwargs) if first else {}
|
||||
if embeds:
|
||||
extra["embeds"] = embeds
|
||||
return await destination.send(content=content, **extra)
|
||||
@@ -0,0 +1,124 @@
|
||||
"""text / formatting helpers: chunking over the message cap, monospace tables, discord
|
||||
dynamic timestamps, and human-readable durations, plus the shared link-wrap / truncate
|
||||
primitives the embed helpers reuse"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from datetime import datetime, timezone
|
||||
from typing import Optional, Sequence
|
||||
|
||||
from .limits import MSG_LIMIT
|
||||
|
||||
_ELLIPSIS = "…"
|
||||
_TIMESTAMP_STYLES = ("t", "T", "d", "D", "f", "F", "R")
|
||||
|
||||
_URL_RE = re.compile(r"https?://[^\s<>]+", re.IGNORECASE)
|
||||
_DISCORD_HOST_RE = re.compile(
|
||||
r"^https?://(?:[a-z0-9-]+\.)*discord(?:app)?\.(?:com|gg|media)\b", re.IGNORECASE
|
||||
)
|
||||
|
||||
|
||||
def truncate(text: str, limit: int) -> str:
|
||||
"""return text clipped to limit chars, appending an ellipsis where it cut"""
|
||||
if len(text) <= limit:
|
||||
return text
|
||||
if limit <= 1:
|
||||
return _ELLIPSIS[:limit]
|
||||
return text[: limit - 1] + _ELLIPSIS
|
||||
|
||||
|
||||
def wrap_bare_links(text: str) -> str:
|
||||
"""wrap non-discord http(s) links in <> so discord does not unfurl them; discord links
|
||||
(which the client renders specially) are left bare, and links already inside <> are not
|
||||
double-wrapped"""
|
||||
if not text:
|
||||
return text
|
||||
|
||||
def repl(match: "re.Match[str]") -> str:
|
||||
url = match.group(0)
|
||||
start = match.start()
|
||||
if start > 0 and text[start - 1] == "<":
|
||||
return url
|
||||
if _DISCORD_HOST_RE.match(url):
|
||||
return url
|
||||
return f"<{url}>"
|
||||
|
||||
return _URL_RE.sub(repl, text)
|
||||
|
||||
|
||||
def chunk_text(text: str, limit: int = MSG_LIMIT) -> list[str]:
|
||||
"""split text into pieces each <= limit, breaking on newline boundaries where possible,
|
||||
then word boundaries, only mid-word as a last resort; no content is lost"""
|
||||
if limit < 1:
|
||||
raise ValueError(f"limit must be >= 1, got {limit}")
|
||||
if len(text) <= limit:
|
||||
return [text] if text else []
|
||||
|
||||
chunks: list[str] = []
|
||||
remaining = text
|
||||
while len(remaining) > limit:
|
||||
window = remaining[:limit]
|
||||
split = window.rfind("\n")
|
||||
if split <= 0:
|
||||
split = window.rfind(" ")
|
||||
if split <= 0:
|
||||
split = limit
|
||||
chunks.append(remaining[:split])
|
||||
remaining = remaining[split:]
|
||||
if remaining.startswith(("\n", " ")):
|
||||
remaining = remaining[1:]
|
||||
if remaining:
|
||||
chunks.append(remaining)
|
||||
return chunks
|
||||
|
||||
|
||||
def format_table(rows: "Sequence[Sequence]", headers: "Optional[Sequence]" = None) -> str:
|
||||
"""render rows as an aligned monospace table wrapped in a code block, column widths
|
||||
auto-sized; raises ValueError on ragged rows"""
|
||||
matrix = [[str(cell) for cell in row] for row in rows]
|
||||
width = len(headers) if headers is not None else (len(matrix[0]) if matrix else 0)
|
||||
for i, row in enumerate(matrix):
|
||||
if len(row) != width:
|
||||
raise ValueError(f"row {i} has {len(row)} cells, expected {width}")
|
||||
|
||||
head = [str(h) for h in headers] if headers is not None else None
|
||||
widths = [0] * width
|
||||
for row in ([head] if head else []) + matrix:
|
||||
for c, cell in enumerate(row):
|
||||
widths[c] = max(widths[c], len(cell))
|
||||
|
||||
def render(row: "list[str]") -> str:
|
||||
return " ".join(cell.ljust(widths[c]) for c, cell in enumerate(row)).rstrip()
|
||||
|
||||
lines = []
|
||||
if head:
|
||||
lines.append(render(head))
|
||||
lines.append(" ".join("-" * widths[c] for c in range(width)))
|
||||
lines.extend(render(row) for row in matrix)
|
||||
return "```\n" + "\n".join(lines) + "\n```"
|
||||
|
||||
|
||||
def discord_timestamp(dt: datetime, style: str = "f") -> str:
|
||||
"""return discord's dynamic timestamp markup <t:unix:style> for a datetime; raises
|
||||
ValueError on an invalid style (one of t/T/d/D/f/F/R)"""
|
||||
if style not in _TIMESTAMP_STYLES:
|
||||
raise ValueError(f"invalid timestamp style '{style}'; expected one of {_TIMESTAMP_STYLES}")
|
||||
if dt.tzinfo is None:
|
||||
dt = dt.replace(tzinfo=timezone.utc)
|
||||
return f"<t:{int(dt.timestamp())}:{style}>"
|
||||
|
||||
|
||||
def humanize_delta(seconds: float) -> str:
|
||||
"""human-readable duration ('2h 5m', '3d 4h') from a seconds count; shows the two most
|
||||
significant non-zero units, or '0s' for zero"""
|
||||
total = int(abs(seconds))
|
||||
sign = "-" if seconds < 0 else ""
|
||||
units = (("d", 86400), ("h", 3600), ("m", 60), ("s", 1))
|
||||
parts = []
|
||||
for label, size in units:
|
||||
if total >= size:
|
||||
value, total = divmod(total, size)
|
||||
parts.append(f"{value}{label}")
|
||||
if not parts:
|
||||
return "0s"
|
||||
return sign + " ".join(parts[:2])
|
||||
Reference in New Issue
Block a user