ad3laid3/disco-reaper
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

improve permission error notice

Browse source
This commit is contained in:
rambros 2026-02-22 17:46:43 +05:30
parent c6ee89742f
commit c5abc6d0e5
4 changed files with 155 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
src/core/engine.py
View file

@ -37,10 +37,13 @@ class MigrationEngine:
"discord_bot_name": d_valid.get("bot_name"),
"discord_server": d_valid.get("server", False),
"discord_server_name": d_valid.get("server_name"),
"discord_intents": d_valid.get("intents", {}),
"discord_permissions": d_valid.get("permissions", {}),
"fluxer_token": f_valid.get("token", False),
"fluxer_bot_name": f_valid.get("bot_name"),
"fluxer_community": f_valid.get("community", False),
"fluxer_community_name": f_valid.get("community_name")
"fluxer_community_name": f_valid.get("community_name"),
"fluxer_permissions": f_valid.get("permissions", {})
}
except Exception as e:
logger.error(f"Validation failed with exception: {e}")

25
src/discord_bot/reader.py
View file

@ -26,8 +26,15 @@ class DiscordReader:
self.guild = await self.client.fetch_guild(self.server_id)
async def validate(self) -> Dict[str, Any]:
"""Validates the token and server ID."""
results = {"token": False, "server": False, "bot_name": None, "server_name": None}
"""Validates the token, server ID, intents, and permissions."""
results = {
"token": False,
"server": False,
"bot_name": None,
"server_name": None,
"intents": {"message_content": False},
"permissions": {"view_channel": False, "read_message_history": False}
}
temp_client = self._create_client()
try:
await temp_client.login(self.token)
@ -39,6 +46,20 @@ class DiscordReader:
if guild is not None:
results["server"] = True
results["server_name"] = guild.name
# Check intents
results["intents"]["message_content"] = temp_client.intents.message_content
# Check permissions
# We need to fetch the member to check permissions
try:
member = await guild.fetch_member(temp_client.user.id)
perms = member.guild_permissions
results["permissions"]["view_channel"] = perms.view_channel
results["permissions"]["read_message_history"] = perms.read_message_history
except Exception:
# Fallback if member fetch fails, though it shouldn't for the bot itself
pass
except Exception:
pass
finally:

64
src/fluxer_bot/writer.py
View file

@ -68,7 +68,7 @@ class FluxerWriter:
return self.bot._http if self.bot else None
async def validate(self) -> dict:
"""Validates the token and community ID."""
"""Validates the token, community ID, and permissions."""
if not self.bot or not self._ready_event.is_set():
await self.start()
@ -76,23 +76,74 @@ class FluxerWriter:
is_community_valid = False
bot_name = None
community_name = None
permissions = {
"manage_channels": False,
"manage_messages": False,
"manage_roles": False,
"manage_emojis_stickers": False,
"manage_webhooks": False
}
try:
# Check token by fetching me
me_id = None
if self.bot and self.bot.user:
is_token_valid = True
bot_name = self.bot.user.username
me_id = self.bot.user.id
else:
# Fallback if on_ready didn't fire yet but we want to check token
me = await self.client.get_current_user()
if me:
is_token_valid = True
bot_name = me.get("username")
me_id = int(me["id"])
# Check community
guild = await self.client.get_guild(self.community_id)
if guild:
guild_data = await self.client.get_guild(self.community_id)
if guild_data:
is_community_valid = True
community_name = guild.get("name")
community_name = guild_data.get("name")
if me_id:
try:
# Fetch member to get roles
member_data = await self.client.get_guild_member(self.community_id, me_id)
member_role_ids = [int(r) for r in member_data.get("roles", [])]
# Fetch all roles to get their permissions
all_roles_data = await self.client.get_guild_roles(self.community_id)
# Calculate total permissions
# In Discord/Fluxer, permissions are additive
total_perms = 0
for r_data in all_roles_data:
r_id = int(r_data["id"])
# Add @everyone permissions (role ID same as guild ID)
if r_id == int(self.community_id) or r_id in member_role_ids:
total_perms |= int(r_data.get("permissions", 0))
# Debugging
# print(f"DEBUG: me_id={me_id}, roles={member_role_ids}, total_perms={total_perms}")
# Bitmask Mapping (Discord standard)
# Administrator: 1 << 3
# Manage Channels: 1 << 4
# Manage Messages: 1 << 13
# Manage Roles: 1 << 28
# Manage Webhooks: 1 << 29
# Manage Emojis/Stickers: 1 << 30
is_admin = bool(total_perms & (1 << 3))
permissions["manage_channels"] = is_admin or bool(total_perms & (1 << 4))
permissions["manage_messages"] = is_admin or bool(total_perms & (1 << 13))
permissions["manage_roles"] = is_admin or bool(total_perms & (1 << 28))
permissions["manage_webhooks"] = is_admin or bool(total_perms & (1 << 29))
permissions["manage_emojis_stickers"] = is_admin or bool(total_perms & (1 << 30))
if is_admin:
logger.info(f"Fluxer bot {bot_name} has Administrator permission.")
except Exception as e:
logger.error(f"Failed to calculate Fluxer permissions: {e}")
except Exception:
pass
@ -100,7 +151,8 @@ class FluxerWriter:
"token": is_token_valid,
"community": is_community_valid,
"bot_name": bot_name,
"community_name": community_name
"community_name": community_name,
"permissions": permissions
}
async def create_channel(self, name: str, topic: str = "", type: int = 0, parent_id: Optional[str] = None) -> str:

73
src/ui/app.py
View file

@ -68,8 +68,10 @@ class MigrationCLI:
self.validation_results = {
"discord_token": False, "discord_bot_name": None,
"discord_server": False, "discord_server_name": None,
"discord_intents": {}, "discord_permissions": {},
"fluxer_token": False, "fluxer_bot_name": None,
"fluxer_community": False, "fluxer_community_name": None
"fluxer_community": False, "fluxer_community_name": None,
"fluxer_permissions": {}
}
self.tokens_valid = False
@ -97,6 +99,11 @@ class MigrationCLI:
console.print("[bold red]Discord Token validation failed (Invalid Token).[/bold red]")
elif not res.get("server"):
console.print("[bold red]Discord Server ID validation failed (Invalid/Inaccessible ID).[/bold red]")
else:
# Store intents & permissions for later UI display
self.validation_results["discord_intents"] = res.get("intents", {})
self.validation_results["discord_permissions"] = res.get("permissions", {})
except Exception as e:
console.print(f"[bold red]Discord validation failed with error: {e}[/bold red]")
else:
@ -115,13 +122,37 @@ class MigrationCLI:
console.print("[bold red]Fluxer Token validation failed (Invalid Token).[/bold red]")
elif not res.get("community"):
console.print("[bold red]Fluxer Community ID validation failed (Invalid/Inaccessible ID).[/bold red]")
else:
# Store permissions
self.validation_results["fluxer_permissions"] = res.get("permissions", {})
except Exception as e:
console.print(f"[bold red]Fluxer validation failed with error: {e}[/bold red]")
else:
console.print("[bold red]Fluxer bot token validation timed out after 10 seconds.[/bold red]")
fluxer_task.cancel()
self.tokens_valid = all(self.validation_results.values())
# Only tokens and server/community existence are strictly required for 'tokens_valid'
self.tokens_valid = (
self.validation_results.get("discord_token") and
self.validation_results.get("discord_server") and
self.validation_results.get("fluxer_token") and
self.validation_results.get("fluxer_community")
)
# Check if all permissions are actually granted
self.permissions_complete = True
if self.tokens_valid:
# Discord
d_intents = self.validation_results.get("discord_intents", {})
d_perms = self.validation_results.get("discord_permissions", {})
if not all([d_intents.get("message_content"), d_perms.get("view_channel"), d_perms.get("read_message_history")]):
self.permissions_complete = False
# Fluxer
f_perms = self.validation_results.get("fluxer_permissions", {})
if not all(f_perms.values()) if f_perms else True:
self.permissions_complete = False
except Exception as e:
console.print(f"[bold red]Validation system failure: {e}[/bold red]")
@ -151,7 +182,13 @@ class MigrationCLI:
console.print("(4) Sync Server Name, Logo and Banner")
console.print("(5) Migrate message history")
val_status = "[bold green][VALID][/bold green]" if self.tokens_valid else "[bold red][INVALID][/bold red]"
if not self.tokens_valid:
val_status = "[bold red][INVALID][/bold red]"
elif not self.permissions_complete:
val_status = "[bold yellow][PERMISSION MISSING][/bold yellow]"
else:
val_status = "[bold green][VALID][/bold green]"
console.print(f"(6) Configuration {val_status}")
console.print("(7) [bold red]⚠ Danger Zone[/bold red]")
@ -180,6 +217,34 @@ class MigrationCLI:
async def edit_configuration(self):
await self.validate_config()
# Display Required Permissions FIRST
def fmt(name, val):
if val is None: return f"[dim]{name}[/dim]" # Unknown
return f"[green]{name}[/green]" if val else f"[red]{name} [MISSING][/red]"
d_intents = self.validation_results.get("discord_intents", {})
d_perms = self.validation_results.get("discord_permissions", {})
f_perms = self.validation_results.get("fluxer_permissions", {})
perm_table = Table(show_header=True, header_style="bold magenta", box=None, padding=(0, 2))
perm_table.add_column("Bot Type")
perm_table.add_column("Required Permissions & Intents")
perm_table.add_row(
"[bold #5865F2]Discord Bot[/bold #5865F2]",
f"• [bold]Intents:[/bold] Server Members, {fmt('Message Content', d_intents.get('message_content'))}\n"
f"• [bold]Permissions:[/bold] {fmt('View Channels', d_perms.get('view_channel'))}, {fmt('Read Message History', d_perms.get('read_message_history'))}"
)
perm_table.add_row(
"[bold #4641D9]Fluxer Bot[/bold #4641D9]",
f"• [bold]Permissions:[/bold] {fmt('Manage Channels', f_perms.get('manage_channels'))}, {fmt('Manage Messages', f_perms.get('manage_messages'))},\n"
f" {fmt('Manage Roles', f_perms.get('manage_roles'))}, {fmt('Manage Emojis/Stickers', f_perms.get('manage_emojis_stickers'))}, {fmt('Manage Webhooks', f_perms.get('manage_webhooks'))}"
)
console.print("\n") # Add spacing before panel
console.print(Panel(perm_table, title="[bold]Required Bot Permissions[/bold]", expand=False, border_style="dim"))
console.print("\n[bold]Configuration Status:[/bold]")
def get_status_str(is_valid, name=None):
@ -193,7 +258,7 @@ class MigrationCLI:
console.print(f"Discord Server ID {get_status_str(self.validation_results.get('discord_server', False), self.validation_results.get('discord_server_name'))}")
console.print(f"Fluxer Community ID {get_status_str(self.validation_results.get('fluxer_community', False), self.validation_results.get('fluxer_community_name'))}")
if not Confirm.ask("Edit now?"):
if not Confirm.ask("\nEdit now?"):
return
console.print("\n[bold]Configuration Editor[/bold] (leave blank to keep current)")