SheWhoReignsSupreme/EveCogs
1
0
Fork 0
Code Issues 4 Pull requests Projects Releases Packages Wiki Activity Actions

feat: add UserSlash cog — user-installable slash commands for every Red command

Browse source 
Replaces OneTrueSlash with full user-install support:
- Single proxy slash command bridges all text commands
- Works in servers, DMs, and group DMs
- REST API patcher sets integration_types & contexts on all global commands
- Owner commands: [p]userslash sync/patch/status
- Fuzzy autocomplete (rapidfuzz)
- Enhanced InterContext/InterMessage for non-guild contexts
This commit is contained in:
everestmcarthur 2026-05-14 20:14:33 +00:00
parent 004d1d61ea
commit 8858470392
9 changed files with 1147 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
info.json
View file

@ -3,6 +3,6 @@
"install_msg": "Welcome to **EveCogs**! Use `[p]cog list` to see available cogs.",
"name": "EveCogs",
"short": "A collection of advanced cogs for Red-DiscordBot.",
"description": "EveCogs is a premium cog repository featuring NexusCore (the ultimate all-in-one server management cog with tickets, applications, suggestions, reaction roles, giveaways, logging, moderation & economy), Wormhole v3.4.0 (cross-server relay with 85+ hybrid commands, context menus, ToS gate, reports, analytics & more), NewHelpMenu (Components V2 help overhaul), VoiceMeister (ultimate temporary voice channel system with Join-to-Create, persistent control panel, name templates, game detection & logging), SelfbotGuard (advanced selfbot detection), and ServerWhitelist (owner-only server join whitelist with blacklist, lock mode, interactive browser, purge & export).",
"description": "EveCogs is a premium cog repository featuring NexusCore (the ultimate all-in-one server management cog with tickets, applications, suggestions, reaction roles, giveaways, logging, moderation & economy), Wormhole v3.4.0 (cross-server relay with 85+ hybrid commands, context menus, ToS gate, reports, analytics & more), NewHelpMenu (Components V2 help overhaul), VoiceMeister (ultimate temporary voice channel system with Join-to-Create, persistent control panel, name templates, game detection & logging), SelfbotGuard (advanced selfbot detection), ServerWhitelist (owner-only server join whitelist with blacklist, lock mode, interactive browser, purge & export), and UserSlash (user-installable slash-command bridge — every Red command works in servers, DMs & group DMs).",
"type": "REPO"
}

240
userslash/__init__.py Normal file
View file

@ -0,0 +1,240 @@
"""
UserSlash — User-installable slash-command bridge for Red-DiscordBot.
Replaces OneTrueSlash. Registers a single proxy slash command (named after
the bot) that makes *every* Red text command accessible via slash — in
servers, DMs, and group DMs. After syncing, all global commands are
automatically patched with ``integration_types: [0, 1]`` and
``contexts: [0, 1, 2]`` so the bot works as a user-installable app.
Owner commands:
[p]userslash sync — Sync the command tree + patch user-install flags
[p]userslash patch — Patch flags on already-synced commands
[p]userslash status — Show per-command user-install status
"""
import asyncio
import logging
from typing import Optional
import discord
from redbot.core import app_commands, commands
from redbot.core.bot import Red
from redbot.core.errors import CogLoadError
from .commands import user_slash_command
from .context import InterContext
from .patcher import apply_user_install_native, patch_all_commands
from .utils import valid_app_name
LOG = logging.getLogger("red.evecogs.userslash")
__red_end_user_data_statement__ = (
"This cog does not persistently store any data or metadata about users."
)
class UserSlash(commands.Cog):
"""User-installable slash commands for every Red command."""
def __init__(self, bot: Red):
self.bot = bot
self._slash_registered = False
# ------------------------------------------------------------------
# Lifecycle
# ------------------------------------------------------------------
async def cog_load(self) -> None:
self.bot.before_invoke(self._before_hook)
self.bot.add_dev_env_value(
"interaction", lambda ctx: getattr(ctx, "interaction", None)
)
asyncio.create_task(self._delayed_setup()) # noqa: RUF006
async def cog_unload(self) -> None:
self.bot.remove_before_invoke_hook(self._before_hook)
try:
self.bot.remove_dev_env_value("interaction")
except Exception:
pass
if self._slash_registered:
self.bot.tree.remove_command(user_slash_command.name, guild=None)
async def _delayed_setup(self) -> None:
"""Wait for the bot to be fully ready, then register the slash command."""
await self.bot.wait_until_red_ready()
assert self.bot.user
# Apply native user-install flags (discord.py 2.4+)
apply_user_install_native(user_slash_command)
try:
user_slash_command.name = valid_app_name(self.bot.user.name)
self.bot.tree.add_command(user_slash_command, guild=None)
self._slash_registered = True
LOG.info(
"Registered /%s as user-installable proxy command.",
user_slash_command.name,
)
except ValueError:
await self.bot.send_to_owners(
f"`userslash` could not convert the name {self.bot.user.name!r} "
"into a valid slash-command name. The name was left unchanged."
)
except app_commands.CommandAlreadyRegistered:
raise CogLoadError(
f"A slash command named `{user_slash_command.name}` is already "
"registered. Unload **OneTrueSlash** (or the conflicting cog) first."
) from None
except app_commands.CommandLimitReached:
raise CogLoadError(
f"{self.bot.user.name} has reached the 100-command global limit."
) from None
# ------------------------------------------------------------------
# Hooks & listeners
# ------------------------------------------------------------------
@staticmethod
async def _before_hook(ctx: commands.Context) -> None:
"""Attach the real Interaction to the context so hybrid commands work."""
interaction: Optional[discord.Interaction] = getattr(
ctx, "_interaction", None
)
if not interaction or getattr(ctx.command, "__commands_is_hybrid__", False):
return
ctx.interaction = interaction # type: ignore[attr-defined]
if not interaction.response.is_done():
ctx._deferring = True # type: ignore[attr-defined]
await interaction.response.defer(ephemeral=False)
@commands.Cog.listener()
async def on_user_update(
self, before: discord.User, after: discord.User
) -> None:
"""Update the slash-command name when the bot's username changes."""
assert self.bot.user
if after.id != self.bot.user.id or before.name == after.name:
return
old_name = user_slash_command.name
try:
user_slash_command.name = valid_app_name(after.name)
except ValueError:
await self.bot.send_to_owners(
f"`userslash` could not convert {after.name!r} into a valid "
"slash-command name. The name was left unchanged."
)
return
self.bot.tree.remove_command(old_name)
apply_user_install_native(user_slash_command)
self.bot.tree.add_command(user_slash_command, guild=None)
await self.bot.send_to_owners(
"The bot's username changed. `userslash`'s slash command has been "
"updated.\n**Run `[p]userslash sync` to push the change to Discord.**"
)
# ------------------------------------------------------------------
# Owner commands
# ------------------------------------------------------------------
@commands.is_owner()
@commands.group(name="userslash")
async def _userslash(self, ctx: commands.Context) -> None:
"""Manage UserSlash — the user-installable slash bridge."""
if ctx.invoked_subcommand is None:
await ctx.send_help(ctx.command)
@_userslash.command(name="sync")
async def _sync(self, ctx: commands.Context) -> None:
"""Sync the command tree, then patch every command for user-install."""
msg = await ctx.send("⏳ Syncing command tree…")
try:
synced = await self.bot.tree.sync()
except discord.HTTPException as exc:
await msg.edit(content=f"❌ Sync failed: {exc}")
return
await msg.edit(
content=f"✅ Synced **{len(synced)}** commands. Patching user-install flags…"
)
try:
patched, total = await patch_all_commands(self.bot)
await msg.edit(
content=(
f"✅ **Done!** Synced {len(synced)} commands — "
f"patched {patched}/{total} with user-install flags.\n"
"Users can now install the bot to their profile and use "
"commands in DMs & group DMs."
)
)
except discord.HTTPException as exc:
await msg.edit(
content=(
f"⚠️ Synced {len(synced)} commands but patching failed: {exc}\n"
"Try `[p]userslash patch` again in a moment."
)
)
@_userslash.command(name="patch")
async def _patch(self, ctx: commands.Context) -> None:
"""Patch already-registered commands with user-install flags (no full sync)."""
msg = await ctx.send("⏳ Patching commands…")
try:
patched, total = await patch_all_commands(self.bot)
await msg.edit(
content=f"✅ Patched **{patched}/{total}** commands with user-install flags."
)
except discord.HTTPException as exc:
await msg.edit(content=f"❌ Patching failed: {exc}")
@_userslash.command(name="status")
async def _status(self, ctx: commands.Context) -> None:
"""Show user-install status for every registered global command."""
try:
from .patcher import get_global_commands
cmds = await get_global_commands(self.bot)
except Exception as exc:
await ctx.send(f"❌ Could not fetch commands: {exc}")
return
if not cmds:
await ctx.send("No global commands are registered.")
return
lines = []
for cmd in cmds:
types = cmd.get("integration_types", [0])
ctxs = cmd.get("contexts") or [0]
user_ok = 1 in types
dm_ok = 1 in ctxs
gdm_ok = 2 in ctxs
ok = user_ok and dm_ok and gdm_ok
lines.append(
f"{'✅' if ok else '⚠️'} `/{cmd['name']}` — "
f"User: {'Yes' if user_ok else 'No'} · "
f"DM: {'Yes' if dm_ok else 'No'} · "
f"Group: {'Yes' if gdm_ok else 'No'}"
)
# Paginate if needed (2000 char limit)
page = []
length = 0
for line in lines:
if length + len(line) + 1 > 1900:
await ctx.send("\n".join(page))
page = []
length = 0
page.append(line)
length += len(line) + 1
if page:
await ctx.send("\n".join(page))
async def setup(bot: Red) -> None:
await bot.add_cog(UserSlash(bot))

50
userslash/channel.py Normal file
View file

@ -0,0 +1,50 @@
"""
InterChannel — thin overlay that routes ``send`` / ``typing`` / ``permissions_for``
through the slash-interaction context instead of the regular channel object.
"""
from typing import TYPE_CHECKING, Union, cast
import discord
from .utils import Thinking, contexts
if TYPE_CHECKING:
from discord.context_managers import Typing
Base = discord.abc.Messageable
else:
Base = object
class InterChannel(Base):
"""Mixed into the real channel class at runtime so existing code keeps working."""
__slots__ = ()
def permissions_for(
self, obj: Union[discord.abc.User, discord.Role], /
) -> discord.Permissions:
try:
ctx = contexts.get()
except LookupError:
pass
else:
interaction = ctx._interaction
bot_user = cast(discord.ClientUser, ctx.bot.user)
if obj.id == interaction.user.id:
return ctx.permissions
elif obj.id == bot_user.id:
return ctx.bot_permissions
return super().permissions_for(obj) # type: ignore
def send(self, *args, **kwargs):
return contexts.get(super()).send(*args, **kwargs)
def typing(self) -> Union[Thinking, "Typing"]:
try:
ctx = contexts.get()
except LookupError:
return super().typing()
else:
return Thinking(ctx)

237
userslash/commands.py Normal file
View file

@ -0,0 +1,237 @@
"""
The proxy slash command and its autocomplete / error handler.
This is the heart of UserSlash: a single ``/botname`` command that bridges
*every* Red text command through Discord's slash-command interface, with
full support for DM / group-DM / non-member-guild contexts.
"""
import asyncio
import functools
import heapq
import logging
import operator
from copy import copy
from typing import Awaitable, Callable, Dict, List, Optional, Tuple, cast
import discord
from rapidfuzz import fuzz
from redbot.core import app_commands, commands
from redbot.core.bot import Red
from redbot.core.commands.help import HelpSettings
from redbot.core.i18n import set_contextual_locale
from .context import InterContext
from .utils import walk_aliases
LOG = logging.getLogger("red.evecogs.userslash.commands")
# ------------------------------------------------------------------
# The one true (user-installable) slash command
# ------------------------------------------------------------------
@app_commands.command(extras={"red_force_enable": True})
async def user_slash_command(
interaction: discord.Interaction,
command: str,
arguments: Optional[str] = None,
attachment: Optional[discord.Attachment] = None,
) -> None:
"""
Run any bot command from anywhere — servers, DMs, and group DMs.
Parameters
-----------
command: str
The text-based command to run.
arguments: Optional[str]
The arguments to provide to the command, if any.
attachment: Optional[Attachment]
An attached file to provide to the command, if any.
"""
assert isinstance(interaction.client, Red)
set_contextual_locale(str(interaction.guild_locale or interaction.locale))
actual = interaction.client.get_command(command)
ctx = await InterContext.from_interaction(interaction, recreate_message=True)
error = None
if command == "help":
ctx._deferring = True
await interaction.response.defer(ephemeral=True)
actual = None
if arguments:
actual = interaction.client.get_command(arguments)
if actual and (signature := actual.signature):
actual = copy(actual)
actual.usage = f"arguments:{signature}"
await interaction.client.send_help_for(
ctx, actual or interaction.client, from_help_command=True
)
else:
ferror: asyncio.Task[Tuple[InterContext, commands.CommandError]] = (
asyncio.create_task(
interaction.client.wait_for(
"command_error", check=lambda c, _: c is ctx
)
)
)
ferror.add_done_callback(
lambda _: setattr(ctx, "interaction", interaction)
)
await interaction.client.invoke(ctx)
if not interaction.response.is_done():
ctx._deferring = True
await interaction.response.defer(ephemeral=True)
if ferror.done():
error = ferror.exception() or ferror.result()[1]
ferror.cancel()
# ---- Followup / cleanup ----
if ctx._deferring and not interaction.is_expired():
if error is None:
if ctx._ticked:
await interaction.followup.send(ctx._ticked, ephemeral=True)
else:
await interaction.delete_original_response()
elif isinstance(error, commands.CommandNotFound):
await interaction.followup.send(
f"❌ Command `{command}` was not found.", ephemeral=True
)
elif isinstance(error, commands.NoPrivateMessage):
await interaction.followup.send(
f"❌ `{command}` can only be used inside a server.",
ephemeral=True,
)
elif isinstance(error, commands.CheckFailure):
if not interaction.guild:
await interaction.followup.send(
f"❌ `{command}` requires a server context or you lack the "
"needed permissions.",
ephemeral=True,
)
else:
await interaction.followup.send(
f"❌ You don't have permission to run `{command}`.",
ephemeral=True,
)
# ------------------------------------------------------------------
# Autocomplete — fuzzy-matches available commands
# ------------------------------------------------------------------
@user_slash_command.autocomplete("command")
async def _autocomplete(
interaction: discord.Interaction, current: str
) -> List[app_commands.Choice[str]]:
assert isinstance(interaction.client, Red)
try:
if not await interaction.client.allowed_by_whitelist_blacklist(
interaction.user
):
return []
except Exception:
pass # safety net for unusual DM contexts
try:
ctx = await InterContext.from_interaction(interaction)
eligible = await interaction.client.message_eligible_as_command(
ctx.message
)
except Exception:
eligible = True # slash interactions are always "eligible"
ctx = await InterContext.from_interaction(interaction)
if not eligible:
# Still provide basic help for slash
return [app_commands.Choice(name="help", value="help")]
try:
help_settings = await HelpSettings.from_context(ctx)
except Exception:
# Fallback settings when guild config is unavailable
help_settings = await HelpSettings.from_context.__wrapped__(ctx) # type: ignore
if current:
extracted = cast(
List[str],
await asyncio.get_event_loop().run_in_executor(
None,
heapq.nlargest,
6,
walk_aliases(
interaction.client, show_hidden=help_settings.show_hidden
),
functools.partial(fuzz.token_sort_ratio, current),
),
)
extracted.append("help")
else:
extracted = ["help"]
_filter: Callable[[commands.Command], Awaitable[bool]] = (
operator.methodcaller(
"can_run" if help_settings.show_hidden else "can_see", ctx
)
)
matches: Dict[commands.Command, str] = {}
for name in extracted:
cmd = interaction.client.get_command(name)
if not cmd or cmd in matches:
continue
try:
if (name == "help" and await cmd.can_run(ctx)) or await _filter(cmd):
if len(name) > 100:
name = name[:99] + "\N{HORIZONTAL ELLIPSIS}"
matches[cmd] = name
except commands.CommandError:
pass
except Exception:
# Swallow unexpected errors in DM/group-DM contexts so autocomplete
# still returns *something*.
pass
return [app_commands.Choice(name=n, value=n) for n in matches.values()]
# ------------------------------------------------------------------
# Error handler
# ------------------------------------------------------------------
@user_slash_command.error
async def _error(interaction: discord.Interaction, error: Exception):
assert isinstance(interaction.client, Red)
if isinstance(error, app_commands.CommandInvokeError):
error = error.original
original = getattr(error, "original", error)
# Friendly message for guild-only commands invoked outside a guild
if isinstance(original, (commands.NoPrivateMessage, commands.CheckFailure)):
try:
if not interaction.response.is_done():
await interaction.response.send_message(
"❌ This command cannot be used here (it may require a server).",
ephemeral=True,
)
else:
await interaction.followup.send(
"❌ This command cannot be used here (it may require a server).",
ephemeral=True,
)
except discord.HTTPException:
pass
return
# Everything else — forward to Red's built-in error handler
ctx = await InterContext.from_interaction(
interaction, recreate_message=True
)
await interaction.client.on_command_error(
ctx, commands.CommandInvokeError(original)
)

215
userslash/context.py Normal file
View file

@ -0,0 +1,215 @@
"""
InterContext — a ``commands.Context`` subclass that bridges slash-command
interactions into Red's text-command pipeline.
Enhanced from OneTrueSlash with robust handling for user-install contexts
where ``guild`` is ``None`` (DMs, group DMs, non-member guilds).
"""
import inspect
import types
from copy import copy
from typing import Optional, Type, Union
import discord
from discord.ext.commands.view import StringView
from redbot.core import commands
from redbot.core.bot import Red
from .message import InterMessage
from .utils import Thinking, contexts
# Parameters that ``discord.abc.Messageable.send`` accepts but
# ``discord.Webhook.send`` does not — we silently discard them.
_INCOMPATIBLE_SEND_PARAMS = tuple(
k
for k in inspect.signature(discord.abc.Messageable.send).parameters
if k not in inspect.signature(discord.Webhook.send).parameters
)
class InterContext(commands.Context):
"""Drop-in ``commands.Context`` replacement for slash interactions."""
_deferring: bool = False
_ticked: Optional[str] = None
_interaction: discord.Interaction[Red]
message: InterMessage
# ------------------------------------------------------------------
# Construction
# ------------------------------------------------------------------
@classmethod
def _get_type(cls, bot: Red) -> Type["InterContext"]:
default = bot.get_context.__kwdefaults__.get("cls", None)
if not isinstance(default, type) or default in cls.__mro__:
return cls
try:
return types.new_class(cls.__name__, (cls, default))
except Exception:
return cls
@classmethod
async def from_interaction(
cls,
interaction: discord.Interaction[Red],
*,
recreate_message: bool = False,
) -> "InterContext":
prefix = f"</{interaction.data['name']}:{interaction.data['id']}> command:"
# Reuse existing context if present on the ContextVar
try:
self = contexts.get()
if recreate_message:
assert self.prefix is not None
self.message._recreate_from_interaction(interaction, prefix)
view = self.view = StringView(self.message.content)
view.skip_string(self.prefix)
invoker = view.get_word()
self.invoked_with = invoker
self.command = interaction.client.all_commands.get(invoker)
return self
except LookupError:
pass
# Build a fresh context
message = InterMessage._from_interaction(interaction, prefix)
view = StringView(message.content)
view.skip_string(prefix)
invoker = view.get_word()
self = cls._get_type(interaction.client)(
message=message,
prefix=prefix,
bot=interaction.client,
view=view,
invoked_with=invoker,
command=interaction.client.all_commands.get(invoker),
)
# Keep _interaction private so d.py parses args the old way
self._interaction = interaction
interaction._baton = self # type: ignore[attr-defined]
contexts.set(self)
return self
# ------------------------------------------------------------------
# Overrides
# ------------------------------------------------------------------
@property
def clean_prefix(self) -> str:
return f"/{self._interaction.data['name']} command:"
async def tick(self, *, message: Optional[str] = None) -> bool:
return await super().tick(message="Done." if message is None else message)
async def react_quietly(
self,
reaction: Union[discord.Emoji, discord.Reaction, discord.PartialEmoji, str],
*,
message: Optional[str] = None,
) -> bool:
self._ticked = f"{reaction} {message}" if message else str(reaction)
return False
async def send(self, *args, **kwargs):
interaction = self._interaction
if interaction.is_expired():
assert interaction.channel
return await interaction.channel.send(*args, **kwargs) # type: ignore
await self.typing(ephemeral=True)
self._deferring = False
delete_after = kwargs.pop("delete_after", None)
for key in _INCOMPATIBLE_SEND_PARAMS:
kwargs.pop(key, None)
m = await interaction.followup.send(*args, **kwargs)
if delete_after:
await m.delete(delay=delete_after)
return m
def typing(self, *, ephemeral: bool = False) -> Thinking:
return Thinking(self, ephemeral=ephemeral)
async def defer(self, *, ephemeral: bool = False) -> None:
await self._interaction.response.defer(ephemeral=ephemeral)
async def send_help(
self,
command: Optional[Union[commands.Command, commands.GroupMixin, str]] = None,
):
command = command or self.command
if isinstance(command, str):
command = self.bot.get_command(command) or command
signature: str
if signature := getattr(command, "signature", ""):
assert not isinstance(command, str)
command = copy(command)
command.usage = f"arguments:{signature}"
return await super().send_help(command)
# ------------------------------------------------------------------
# Permissions (DM-safe)
# ------------------------------------------------------------------
def _apply_implicit_permissions(
self, user: discord.abc.User, base: discord.Permissions
) -> discord.Permissions:
if base.administrator or (
self.guild and self.guild.owner_id == user.id
):
return discord.Permissions.all()
base = copy(base)
if not base.send_messages:
base.send_tts_messages = False
base.mention_everyone = False
base.embed_links = False
base.attach_files = False
if not base.read_messages:
base &= ~discord.Permissions.all_channel()
channel_type = getattr(self.channel, "type", None)
if channel_type in (
discord.ChannelType.voice,
discord.ChannelType.stage_voice,
):
if not base.connect:
denied = discord.Permissions.voice()
denied.update(manage_channels=True, manage_roles=True)
base &= ~denied
elif channel_type not in (
discord.ChannelType.private,
discord.ChannelType.group,
None,
):
# Text channel in a guild — strip voice perms
base &= ~discord.Permissions.voice()
return base
@discord.utils.cached_property
def permissions(self) -> discord.Permissions:
if (
self._interaction._permissions == 0
or not self.guild
):
return discord.Permissions._dm_permissions() # type: ignore
return self._apply_implicit_permissions(
self.author, self._interaction.permissions
)
@discord.utils.cached_property
def bot_permissions(self) -> discord.Permissions:
if not self.guild:
# In DMs the bot can always send, embed, and attach
return discord.Permissions._dm_permissions() | discord.Permissions( # type: ignore
send_messages=True, attach_files=True, embed_links=True
)
return self._apply_implicit_permissions(
self.me, self._interaction.app_permissions
) | discord.Permissions(
send_messages=True, attach_files=True, embed_links=True
)

22
userslash/info.json Normal file
View file

@ -0,0 +1,22 @@
{
"author": ["everestmcarthur"],
"install_msg": "**UserSlash** loaded!\n\n**Setup (one-time):**\n1. Go to the **Discord Developer Portal** → Your Application → **Installation**\n2. Under *Installation Contexts*, enable **User Install**\n3. Under *Default Install Settings — User Install*, add the `applications.commands` scope\n4. Run `[p]userslash sync` to register & patch commands\n5. Share the OAuth2 User-Install link so users can add your bot to their profile!\n\n⚠️ **Unload OneTrueSlash** first if you have it loaded — this cog replaces it.\n\nAll Red text commands are now accessible as slash commands in servers, DMs, and group DMs.",
"name": "UserSlash",
"disabled": false,
"short": "User-installable slash commands — every Red command works in servers, DMs, and group DMs.",
"description": "UserSlash replaces OneTrueSlash with full user-install support. It registers a single proxy slash command (named after the bot) that bridges all Red text commands, then patches every global command with user-install integration types. Commands work everywhere: guild channels, bot DMs, group DMs, and even servers where the bot isn't a member. Includes owner commands for syncing, patching, and checking status. Gracefully handles non-guild contexts with helpful error messages when a command requires a server.",
"tags": [
"slash",
"user-install",
"commands",
"dm",
"group-dm",
"hybrid",
"utility"
],
"type": "COG",
"end_user_data_statement": "This cog does not persistently store any data or metadata about users.",
"min_bot_version": "3.5.0",
"min_python_version": [3, 9, 0],
"requirements": ["rapidfuzz"]
}

195
userslash/message.py Normal file
View file

@ -0,0 +1,195 @@
"""
InterMessage — lightweight ``discord.Message`` stand-in built from
a slash-command ``Interaction`` so Red's text-command pipeline can process it.
Enhanced from OneTrueSlash to handle user-install contexts (DMs, group DMs,
and guilds where the bot is not a member).
"""
import asyncio
from copy import copy
from typing import TypeVar
import discord
from .channel import InterChannel
_TT = TypeVar("_TT", bound=type)
def _noop_step(*_args, **_kwargs):
"""Coroutine that just yields to the event loop once."""
return asyncio.sleep(0)
def neuter_coros(cls: _TT) -> _TT:
"""Replace every inherited coroutine with a harmless no-op."""
for name in dir(cls):
if name in cls.__dict__:
continue
if (attr := getattr(cls, name, None)) is None:
continue
if asyncio.iscoroutinefunction(attr):
setattr(cls, name, property(lambda self: _noop_step))
return cls
@neuter_coros
class InterMessage(discord.Message):
__slots__ = ()
def __init__(self, **kwargs) -> None:
raise RuntimeError("InterMessage must be created via _from_interaction")
# ------------------------------------------------------------------
# Factory
# ------------------------------------------------------------------
@classmethod
def _from_interaction(
cls, interaction: discord.Interaction, prefix: str
) -> "InterMessage":
assert interaction.data
assert interaction.client.user
self = cls.__new__(cls)
self._state = interaction._state
self._edited_timestamp = None
# Constant / default fields
self.tts = False
self.webhook_id = None
self.mention_everyone = False
self.embeds = []
self.role_mentions = []
self.id = interaction.id
self.nonce = None
self.pinned = False
self.type = discord.MessageType.default
self.flags = discord.MessageFlags()
self.reactions = []
self.reference = None
self.application = None
self.activity = None
self.stickers = []
self.components = []
self.role_subscription = None
self.application_id = None
self.position = None
channel = interaction.channel
self.guild = interaction.guild
# --- Determine author & channel based on context ---
if interaction.guild_id and not interaction.guild:
# Guild ID present but guild not in cache (e.g. user-install in a
# server the bot hasn't joined). Treat as DM so Red doesn't crash.
if isinstance(interaction.user, discord.Member):
self.author = interaction.user._user
else:
self.author = interaction.user
channel = cls._make_dm_channel(interaction)
elif not interaction.guild:
# True DM or group DM from a user-install context.
self.author = interaction.user
if channel is None:
channel = cls._make_dm_channel(interaction)
else:
channel = copy(channel)
else:
# Normal guild context — bot is a member.
self.author = interaction.user
if channel is not None:
channel = copy(channel)
else:
channel = cls._make_dm_channel(interaction)
# Mix InterChannel into the channel so send/typing route correctly
if channel is not None:
channel.__class__ = type(
InterChannel.__name__,
(InterChannel, channel.__class__),
{"__slots__": ()},
)
self.channel = channel # type: ignore
self._recreate_from_interaction(interaction, prefix)
return self
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
@staticmethod
def _make_dm_channel(interaction: discord.Interaction) -> discord.DMChannel:
"""Fabricate a minimal DMChannel when the real one is unavailable."""
assert interaction.client.user
user = interaction.user
user_json = (
user._to_minimal_user_json()
if hasattr(user, "_to_minimal_user_json")
else {
"id": str(user.id),
"username": str(user),
"discriminator": "0",
"avatar": None,
}
)
return discord.DMChannel(
me=interaction.client.user,
state=interaction._state,
data={
"id": interaction.channel_id or interaction.id,
"name": f"DM with {user}",
"type": 1,
"last_message_id": None,
"recipients": [
user_json,
interaction.client.user._to_minimal_user_json(),
],
},
)
def _recreate_from_interaction(
self, interaction: discord.Interaction, prefix: str
) -> None:
assert interaction.data and interaction.client.user
self.content = f"{prefix}{interaction.namespace.command}"
if interaction.namespace.arguments:
self.content = f"{self.content} {interaction.namespace.arguments}"
if interaction.namespace.attachment:
self.attachments = [interaction.namespace.attachment]
else:
self.attachments = []
resolved = interaction.data.get("resolved", {})
if self.guild:
self.mentions = [
discord.Member(data=d, guild=self.guild, state=self._state)
for d in resolved.get("members", {}).values()
]
else:
self.mentions = [
discord.User(data=d, state=self._state)
for d in resolved.get("users", {}).values()
]
# ------------------------------------------------------------------
# Overrides
# ------------------------------------------------------------------
def to_reference(self, *, fail_if_not_exists: bool = True):
return None
def to_message_reference_dict(self):
return discord.utils.MISSING
async def reply(self, *args, **kwargs):
return await self.channel.send(*args, **kwargs)
def edit(self, *args, **kwargs):
return asyncio.sleep(0, self)

111
userslash/patcher.py Normal file
View file

@ -0,0 +1,111 @@
"""
REST-API patcher — ensures every global application command carries
``integration_types: [0, 1]`` and ``contexts: [0, 1, 2]`` so the bot
is user-installable and works in guilds, DMs, and group DMs.
"""
import logging
from typing import Any, Dict, List, Tuple
import discord
from redbot.core.bot import Red
LOG = logging.getLogger("red.evecogs.userslash.patcher")
# 0 = GUILD_INSTALL, 1 = USER_INSTALL
INTEGRATION_TYPES: List[int] = [0, 1]
# 0 = GUILD, 1 = BOT_DM, 2 = PRIVATE_CHANNEL (group DMs)
CONTEXTS: List[int] = [0, 1, 2]
# ------------------------------------------------------------------
# Native discord.py 2.4+ helpers (best-effort)
# ------------------------------------------------------------------
def apply_user_install_native(command: Any) -> bool:
"""
Try to set ``allowed_installs`` / ``allowed_contexts`` using the
native discord.py 2.4+ API. Returns ``True`` on success.
"""
try:
from discord.app_commands import AppInstallationType, AppCommandContext
command.allowed_installs = AppInstallationType(guild=True, user=True)
command.allowed_contexts = AppCommandContext(
guild=True, dm_channel=True, private_channel=True
)
return True
except (ImportError, AttributeError, TypeError):
return False
# ------------------------------------------------------------------
# REST fallback — works with any discord.py version
# ------------------------------------------------------------------
async def get_global_commands(bot: Red) -> List[Dict[str, Any]]:
"""Fetch every registered global command from the Discord API."""
app_id = bot.application_id
if not app_id:
raise RuntimeError("Bot application_id is not available yet.")
route = discord.http.Route(
"GET",
"/applications/{application_id}/commands",
application_id=app_id,
)
return await bot.http.request(route)
async def patch_command(
bot: Red,
command_id: int | str,
*,
integration_types: List[int] = INTEGRATION_TYPES,
cmd_contexts: List[int] = CONTEXTS,
) -> Dict[str, Any]:
"""PATCH a single global command with user-install flags."""
route = discord.http.Route(
"PATCH",
"/applications/{application_id}/commands/{command_id}",
application_id=bot.application_id,
command_id=command_id,
)
return await bot.http.request(
route,
json={
"integration_types": integration_types,
"contexts": cmd_contexts,
},
)
async def patch_all_commands(bot: Red) -> Tuple[int, int]:
"""
Patch **all** global commands that are missing user-install flags.
Returns ``(patched_count, total_count)``.
"""
commands = await get_global_commands(bot)
target_types = set(INTEGRATION_TYPES)
target_contexts = set(CONTEXTS)
patched = 0
for cmd in commands:
current_types = set(cmd.get("integration_types", [0]))
current_contexts = set(cmd.get("contexts") or [0])
if current_types == target_types and current_contexts == target_contexts:
continue # already good
try:
await patch_command(bot, cmd["id"])
patched += 1
LOG.info(
"Patched command /%s (%s) with user-install flags.",
cmd["name"],
cmd["id"],
)
except discord.HTTPException as exc:
LOG.warning("Failed to patch /%s: %s", cmd["name"], exc)
return patched, len(commands)

76
userslash/utils.py Normal file
View file

@ -0,0 +1,76 @@
"""
Shared utilities for UserSlash.
"""
from contextvars import ContextVar
from typing import TYPE_CHECKING, Any, Generator, Optional
from discord.ext.commands import GroupMixin
from redbot.core import commands
if TYPE_CHECKING:
from .context import InterContext
try:
import regex as re
except ImportError:
import re
# ContextVar holding the current InterContext for the active interaction
contexts: ContextVar["InterContext"] = ContextVar("userslash_contexts")
def valid_app_name(name: str) -> str:
"""Convert a display name into a valid slash-command name (lowercase, underscores, ≤32 chars)."""
from discord.app_commands.commands import VALID_SLASH_COMMAND_NAME, validate_name
name = "_".join(
re.findall(VALID_SLASH_COMMAND_NAME.pattern.strip("^$"), name.lower())
)
return validate_name(name[:32])
class Thinking:
"""Async context-manager / awaitable that defers the interaction response once."""
def __init__(self, ctx: "InterContext", *, ephemeral: bool = False):
self.ctx = ctx
self.ephemeral = ephemeral
def __await__(self) -> Generator[Any, Any, None]:
ctx = self.ctx
interaction = ctx._interaction
if not ctx._deferring and not interaction.response.is_done():
ctx._deferring = True
return (
yield from interaction.response.defer(
ephemeral=self.ephemeral
).__await__()
)
async def __aenter__(self):
await self
async def __aexit__(self, *args):
pass
def walk_aliases(
group: GroupMixin[Any],
/,
*,
parent: Optional[str] = "",
show_hidden: bool = False,
) -> Generator[str, None, None]:
"""Yield every reachable command name (including aliases) in *group*."""
for name, command in group.all_commands.items():
if command.qualified_name == "help":
continue
if not command.enabled or (not show_hidden and command.hidden):
continue
yield f"{parent}{name}"
if isinstance(command, commands.GroupMixin):
yield from walk_aliases(
command, parent=f"{parent}{name} ", show_hidden=show_hidden
)