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

implement stoat token check

Browse source
This commit is contained in:
rambros 2026-02-24 23:42:43 +05:30
parent 49752229bf
commit 75f7b8ecf9
3 changed files with 80 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
src/core/discord_reader.py
View file

@ -40,7 +40,7 @@ class DiscordReader:
await temp_client.login(self.token)
results["token"] = True
if temp_client.user:
results["bot_name"] = temp_client.user.name
results["bot_name"] = temp_client.user.display_name
guild = await temp_client.fetch_guild(self.server_id)
if guild is not None:

78
src/stoat/writer.py
View file

@ -1,4 +1,5 @@
import logging
import stoat
from typing import Optional, List, Dict, Any
logger = logging.getLogger(__name__)
@ -9,22 +10,77 @@ class StoatWriter:
self.community_id = str(community_id)
async def start(self):
logger.info("StoatWriter start (Not implemented)")
self.client = stoat.Client(token=self.token, bot=True)
async def validate(self) -> dict:
return {
"token": True,
"community": True,
"bot_name": "Stoat Dummy",
"community_name": "Stoat Community Dummy",
results = {
"token": False,
"community": False,
"bot_name": "N/A",
"community_name": "N/A",
"permissions": {
"manage_channels": True,
"manage_messages": True,
"manage_roles": True,
"manage_emojis_stickers": True,
"manage_webhooks": True
"manage_channels": False,
"manage_server": False,
"manage_permissions": False,
"manage_roles": False,
"manage_customization": False
}
}
# Use a temporary client for validation
client = stoat.Client(token=self.token, bot=True)
try:
# Validate token by fetching current user
try:
current_user = await client.fetch_user("@me")
results["token"] = True
results["bot_name"] = current_user.display_name or current_user.name
except stoat.Unauthorized:
logger.error("Invalid Stoat token.")
return results
# Validate server access
try:
server = await client.fetch_server(self.community_id)
results["community"] = True
results["community_name"] = server.name
# Check permissions using effective server permissions for the bot
# Use current_user.id since @me might not be supported in all member endpoints
try:
me = await server.fetch_member(current_user.id)
# We use server.permissions_for(me) instead of me.server_permissions
# to avoid cache-related NoData exceptions.
# safe=False allows calculating even if some roles aren't in local cache.
perms = server.permissions_for(me, safe=False)
results["permissions"] = {
"manage_channels": perms.manage_channels,
"manage_server": perms.manage_server,
"manage_permissions": perms.manage_permissions,
"manage_roles": perms.manage_roles,
"manage_customization": perms.manage_customization
}
except stoat.NotFound:
logger.error(f"Bot member {current_user.id} not found in Stoat server {self.community_id}.")
except stoat.Forbidden:
logger.error(f"Bot lacks permissions to fetch its own member data in Stoat server {self.community_id}.")
except Exception as e:
logger.error(f"Error fetching Stoat member permissions: {e}")
except stoat.NotFound:
logger.error(f"Stoat server {self.community_id} not found.")
except stoat.Forbidden:
logger.error(f"Bot has no access to Stoat server {self.community_id}.")
except Exception as e:
logger.error(f"Error validating Stoat server: {e}")
except Exception as e:
logger.error(f"Stoat validation failed: {str(e)}")
finally:
await client.close()
return results
async def get_channels(self) -> List[Dict[str, Any]]:
return []

19
src/ui/app.py
View file

@ -215,10 +215,15 @@ class MigrationCLI:
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
# Fluxer or Stoat
if self.target_platform == "fluxer":
f_perms = self.validation_results.get("fluxer_permissions", {})
if not all(f_perms.values()) if f_perms else True:
self.permissions_complete = False
else:
s_perms = self.validation_results.get("stoat_permissions", {})
if not all(s_perms.values()) if s_perms else True:
self.permissions_complete = False
except Exception as e:
console.print(f"[bold red]Validation system failure: {e}[/bold red]")
@ -250,7 +255,7 @@ class MigrationCLI:
while True:
console.print("")
console.print(Panel.fit("Fluxer Reaper", style="bold blue"))
console.print(Panel.fit(f"{self.target_platform.capitalize()} Reaper", style="bold blue"))
d_name = self.validation_results.get("discord_server_name")
d_display = f"[bold green]\"{d_name}\"[/bold green]" if d_name else ("[bold yellow]TIMEOUT ERROR[/bold yellow]" if self.validation_results.get("discord_timeout") else "[bold red]NOT SET UP[/bold red]")
@ -338,8 +343,8 @@ class MigrationCLI:
s_perms = self.validation_results.get("stoat_permissions", {})
perm_table.add_row(
"[bold #FF8C00]Stoat Bot[/bold #FF8C00]",
f"• [bold]Permissions:[/bold] {fmt('Manage Channels', s_perms.get('manage_channels'))}, {fmt('Manage Messages', s_perms.get('manage_messages'))},\n"
f" {fmt('Manage Roles', s_perms.get('manage_roles'))}, {fmt('Manage Emojis/Stickers', s_perms.get('manage_emojis_stickers'))}, {fmt('Manage Webhooks', s_perms.get('manage_webhooks'))}"
f"• [bold]Permissions:[/bold] {fmt('Manage Channel', s_perms.get('manage_channels'))}, {fmt('Manage Server', s_perms.get('manage_server'))},\n"
f" {fmt('Manage Permissions', s_perms.get('manage_permissions'))}, {fmt('Manage Roles', s_perms.get('manage_roles'))}, {fmt('Manage Customization', s_perms.get('manage_customization'))}"
)
console.print("\n") # Add spacing before panel