sync role permissions in stoat

This commit is contained in:
rambros 2026-02-26 17:08:08 +05:30
parent 6de9a054f7
commit 9d8f236fb8
3 changed files with 84 additions and 8 deletions

View file

@ -128,6 +128,17 @@ async def sync_permissions(context: MigrationContext, progress_callback: Callabl
async def migrate_roles(context: MigrationContext, progress_callback: Callable[[str, int, int], Awaitable[None]] | None = None, force: bool = False) -> list[str]: async def migrate_roles(context: MigrationContext, progress_callback: Callable[[str, int, int], Awaitable[None]] | None = None, force: bool = False) -> list[str]:
"""Copies roles and their baseline permissions. Returns a list of cloned role names.""" """Copies roles and their baseline permissions. Returns a list of cloned role names."""
# 1. Sync default permissions (@everyone)
try:
guild = context.discord_reader.guild
if guild:
default_role = guild.default_role
await context.stoat_writer.update_default_role_permissions(default_role.permissions.value)
logger.info("Synced default server permissions (@everyone).")
except Exception as e:
logger.error(f"Failed to sync default permissions: {e}")
roles = await context.discord_reader.get_roles() roles = await context.discord_reader.get_roles()
if not force: if not force:
@ -145,7 +156,8 @@ async def migrate_roles(context: MigrationContext, progress_callback: Callable[[
stoat_id = await context.stoat_writer.create_role( stoat_id = await context.stoat_writer.create_role(
name=role.name, name=role.name,
color=role.color.value, color=role.color.value,
hoist=role.hoist hoist=role.hoist,
permissions=role.permissions.value
) )
if stoat_id: if stoat_id:
context.state.set_role_mapping(str(role.id), stoat_id) context.state.set_role_mapping(str(role.id), stoat_id)
@ -155,3 +167,4 @@ async def migrate_roles(context: MigrationContext, progress_callback: Callable[[
await asyncio.sleep(context.config.migration.rate_limit_delay_seconds) await asyncio.sleep(context.config.migration.rate_limit_delay_seconds)
return cloned_role_names return cloned_role_names

View file

@ -294,7 +294,7 @@ class StoatWriter:
return None return None
async def create_role(self, name: str, color: int = 0, hoist: bool = False, **kwargs) -> str: async def create_role(self, name: str, color: int = 0, hoist: bool = False, permissions: int = 0, **kwargs) -> str:
server = await self._get_server() server = await self._get_server()
try: try:
# Create role first (Stoat create_role only takes name and rank) # Create role first (Stoat create_role only takes name and rank)
@ -310,11 +310,74 @@ class StoatWriter:
color=hex_color if hex_color is not None else stoat.UNDEFINED, color=hex_color if hex_color is not None else stoat.UNDEFINED,
hoist=hoist hoist=hoist
) )
# Set permissions
if permissions != 0:
s_perms = self._map_permissions(permissions)
await server.set_role_permissions(role, allow=s_perms)
return str(role.id) return str(role.id)
except Exception as e: except Exception as e:
logger.error(f"Failed to create Stoat role {name}: {e}") logger.error(f"Failed to create Stoat role {name}: {e}")
return "" return ""
def _map_permissions(self, discord_perms_int: int) -> stoat.Permissions:
import discord
d_perms = discord.Permissions(discord_perms_int)
s_perms = stoat.Permissions.none()
mapping = {
"manage_channels": "manage_channels",
"manage_guild": "manage_server",
"manage_roles": "manage_roles",
"kick_members": "kick_members",
"ban_members": "ban_members",
"view_channel": "view_channel",
"send_messages": "send_messages",
"manage_messages": "manage_messages",
"embed_links": "send_embeds",
"attach_files": "upload_files",
"read_message_history": "read_message_history",
"mention_everyone": "mention_everyone",
"add_reactions": "react",
"connect": "connect",
"speak": "speak",
"stream": "video",
"mute_members": "mute_members",
"deafen_members": "deafen_members",
"move_members": "move_members",
"manage_nicknames": "manage_nicknames",
"manage_webhooks": "manage_webhooks",
"manage_emojis": "manage_customization",
"manage_stickers": "manage_customization",
"moderate_members": "timeout_members",
}
for d_name, s_name in mapping.items():
if getattr(d_perms, d_name, False):
try:
setattr(s_perms, s_name, True)
except Exception:
pass
if d_perms.administrator:
return stoat.Permissions.all()
return s_perms
async def update_default_role_permissions(self, permissions: int):
"""Sets the default server permissions from a Discord permissions bitfield."""
server = await self._get_server()
try:
s_perms = self._map_permissions(permissions)
await server.set_default_permissions(s_perms)
return True
except Exception as e:
logger.error(f"Failed to update Stoat default permissions: {e}")
return False
async def create_emoji(self, name: str, image_bytes: bytes, **kwargs) -> str: async def create_emoji(self, name: str, image_bytes: bytes, **kwargs) -> str:
server = await self._get_server() server = await self._get_server()
try: try:
@ -437,9 +500,9 @@ class StoatWriter:
try: try:
channel = await self.client.fetch_channel(channel_id) channel = await self.client.fetch_channel(channel_id)
# Stoat Permissions objects are created from raw integers # Stoat Permissions objects MUST be mapped from Discord bitfields
allow_perms = stoat.Permissions(allow) allow_perms = self._map_permissions(allow)
deny_perms = stoat.Permissions(deny) deny_perms = self._map_permissions(deny)
# If overwrite_id is the community_id, it refers to the default permissions (@everyone) # If overwrite_id is the community_id, it refers to the default permissions (@everyone)
if str(overwrite_id) == self.community_id: if str(overwrite_id) == self.community_id:
@ -449,12 +512,12 @@ class StoatWriter:
# Stoat uses set_role_permissions(role_id, allow=..., deny=...) # Stoat uses set_role_permissions(role_id, allow=..., deny=...)
await channel.set_role_permissions(overwrite_id, allow=allow_perms, deny=deny_perms) await channel.set_role_permissions(overwrite_id, allow=allow_perms, deny=deny_perms)
else: else:
# This case might not be directly supported for individual users in the same way # User-specific overrides are currently skipped for Stoat
# but we'll try something if needed.
pass pass
except Exception as e: except Exception as e:
logger.error(f"Failed to set Stoat channel permission for {overwrite_id} on {channel_id}: {e}") logger.error(f"Failed to set Stoat channel permission for {overwrite_id} on {channel_id}: {e}")
async def delete_all_roles(self, **kwargs) -> int: async def delete_all_roles(self, **kwargs) -> int:
server = await self._get_server() server = await self._get_server()
# Stoat roles are in server.roles (dict) or fetch_roles() # Stoat roles are in server.roles (dict) or fetch_roles()

View file

@ -739,7 +739,7 @@ class MigrationCLI:
role_task = progress.add_task("[cyan]Syncing Roles...", total=100) role_task = progress.add_task("[cyan]Syncing Roles...", total=100)
async def update_progress(item_name: str, current: int, total: int): async def update_progress(item_name: str, current: int, total: int):
progress.update(role_task, total=total, completed=current, description=f"[cyan]Syncing Role: {item_name}") progress.update(role_task, total=total, completed=current, description=f"[cyan]Copying Role: {item_name}")
self.engine.is_running = True self.engine.is_running = True
cloned_roles = await roles_mod.migrate_roles(self.engine, progress_callback=update_progress, force=force) cloned_roles = await roles_mod.migrate_roles(self.engine, progress_callback=update_progress, force=force)