display server,community name in menu

This commit is contained in:
rambros 2026-02-20 13:43:54 +05:30
parent 9da66326fc
commit 5dbcac68b6
4 changed files with 51 additions and 19 deletions

View file

@ -1,6 +1,6 @@
import asyncio import asyncio
import logging import logging
from typing import Callable, Awaitable, List from typing import Callable, Awaitable, List, Dict, Any
from src.config import AppConfig from src.config import AppConfig
from src.core.state import MigrationState from src.core.state import MigrationState
from src.discord_bot.reader import DiscordReader from src.discord_bot.reader import DiscordReader
@ -27,17 +27,20 @@ class MigrationEngine:
self.is_running = False self.is_running = False
async def validate_all(self) -> Dict[str, bool]: async def validate_all(self) -> Dict[str, Any]:
"""Returns True if both connections are valid.""" """Returns True if both connections are valid."""
try: try:
# Note: in real scenarios we might want concurrent validation
d_valid = await self.discord_reader.validate() d_valid = await self.discord_reader.validate()
f_valid = await self.fluxer_writer.validate() f_valid = await self.fluxer_writer.validate()
return { return {
"discord_token": d_valid.get("token", False), "discord_token": d_valid.get("token", False),
"discord_bot_name": d_valid.get("bot_name"),
"discord_server": d_valid.get("server", False), "discord_server": d_valid.get("server", False),
"discord_server_name": d_valid.get("server_name"),
"fluxer_token": f_valid.get("token", False), "fluxer_token": f_valid.get("token", False),
"fluxer_community": f_valid.get("community", False) "fluxer_bot_name": f_valid.get("bot_name"),
"fluxer_community": f_valid.get("community", False),
"fluxer_community_name": f_valid.get("community_name")
} }
except Exception as e: except Exception as e:
logger.error(f"Validation failed with exception: {e}") logger.error(f"Validation failed with exception: {e}")

View file

@ -25,16 +25,20 @@ class DiscordReader:
# In a real app we'd wait until ready, here we simulate fetching the guild # In a real app we'd wait until ready, here we simulate fetching the guild
self.guild = await self.client.fetch_guild(self.server_id) self.guild = await self.client.fetch_guild(self.server_id)
async def validate(self) -> Dict[str, bool]: async def validate(self) -> Dict[str, Any]:
"""Validates the token and server ID.""" """Validates the token and server ID."""
results = {"token": False, "server": False} results = {"token": False, "server": False, "bot_name": None, "server_name": None}
temp_client = self._create_client() temp_client = self._create_client()
try: try:
await temp_client.login(self.token) await temp_client.login(self.token)
results["token"] = True results["token"] = True
if temp_client.user:
results["bot_name"] = temp_client.user.name
guild = await temp_client.fetch_guild(self.server_id) guild = await temp_client.fetch_guild(self.server_id)
if guild is not None: if guild is not None:
results["server"] = True results["server"] = True
results["server_name"] = guild.name
except Exception: except Exception:
pass pass
finally: finally:

View file

@ -21,21 +21,28 @@ class FluxerWriter:
is_token_valid = False is_token_valid = False
is_community_valid = False is_community_valid = False
bot_name = None
community_name = None
try: try:
# Check token by fetching me # Check token by fetching me
await self.client.get_current_user() me = await self.client.get_current_user()
is_token_valid = True if me:
is_token_valid = True
bot_name = me.get("username")
# Check community # Check community
guild = await self.client.get_guild(self.community_id) guild = await self.client.get_guild(self.community_id)
if guild: if guild:
is_community_valid = True is_community_valid = True
community_name = guild.get("name")
except Exception: except Exception:
pass pass
return { return {
"token": is_token_valid, "token": is_token_valid,
"community": is_community_valid "community": is_community_valid,
"bot_name": bot_name,
"community_name": community_name
} }
async def create_channel(self, name: str, topic: str = "", type: int = 0, parent_id: Optional[str] = None) -> str: async def create_channel(self, name: str, topic: str = "", type: int = 0, parent_id: Optional[str] = None) -> str:

View file

@ -55,8 +55,10 @@ class MigrationCLI:
old_settings = termios.tcgetattr(fd) if fd is not None else None old_settings = termios.tcgetattr(fd) if fd is not None else None
self.validation_results = { self.validation_results = {
"discord_token": False, "discord_server": False, "discord_token": False, "discord_bot_name": None,
"fluxer_token": False, "fluxer_community": False "discord_server": False, "discord_server_name": None,
"fluxer_token": False, "fluxer_bot_name": None,
"fluxer_community": False, "fluxer_community_name": None
} }
self.tokens_valid = False self.tokens_valid = False
@ -94,7 +96,9 @@ class MigrationCLI:
try: try:
res = discord_task.result() res = discord_task.result()
self.validation_results["discord_token"] = res.get("token", False) self.validation_results["discord_token"] = res.get("token", False)
self.validation_results["discord_bot_name"] = res.get("bot_name")
self.validation_results["discord_server"] = res.get("server", False) self.validation_results["discord_server"] = res.get("server", False)
self.validation_results["discord_server_name"] = res.get("server_name")
if not res.get("token"): if not res.get("token"):
console.print("[bold red]Discord Token validation failed (Invalid Token).[/bold red]") console.print("[bold red]Discord Token validation failed (Invalid Token).[/bold red]")
elif not res.get("server"): elif not res.get("server"):
@ -110,7 +114,9 @@ class MigrationCLI:
try: try:
res = fluxer_task.result() res = fluxer_task.result()
self.validation_results["fluxer_token"] = res.get("token", False) self.validation_results["fluxer_token"] = res.get("token", False)
self.validation_results["fluxer_bot_name"] = res.get("bot_name")
self.validation_results["fluxer_community"] = res.get("community", False) self.validation_results["fluxer_community"] = res.get("community", False)
self.validation_results["fluxer_community_name"] = res.get("community_name")
if not res.get("token"): if not res.get("token"):
console.print("[bold red]Fluxer Token validation failed (Invalid Token).[/bold red]") console.print("[bold red]Fluxer Token validation failed (Invalid Token).[/bold red]")
elif not res.get("community"): elif not res.get("community"):
@ -146,7 +152,16 @@ class MigrationCLI:
await self.validate_config() await self.validate_config()
while True: while True:
console.print("")
console.print(Panel.fit("Fluxer Reaper", style="bold blue")) console.print(Panel.fit("Fluxer 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 red]NOT SET UP[/bold red]"
f_name = self.validation_results.get("fluxer_community_name")
f_display = f"[bold green]\"{f_name}\"[/bold green]" if f_name else "[bold red]NOT SET UP[/bold red]"
console.print(f"[bold cyan]Discord Server:[/bold cyan] {d_display}")
console.print(f"[bold magenta]Fluxer Community:[/bold magenta] {f_display}")
console.print("[bold]Main Menu[/bold]") console.print("[bold]Main Menu[/bold]")
console.print("(1) Clone Server Template (Channels & Categories)") console.print("(1) Clone Server Template (Channels & Categories)")
console.print("(2) Copy Roles & Permissions") console.print("(2) Copy Roles & Permissions")
@ -181,13 +196,16 @@ class MigrationCLI:
await self.validate_config() await self.validate_config()
console.print("\n[bold]Configuration Status:[/bold]") console.print("\n[bold]Configuration Status:[/bold]")
def get_status_str(is_valid): def get_status_str(is_valid, name=None):
return "[bold green][VALID][/bold green]" if is_valid else "[bold red][INVALID][/bold red]" status = "[bold green][VALID][/bold green]" if is_valid else "[bold red][INVALID][/bold red]"
if is_valid and name:
return f"{status} \"{name}\""
return status
console.print(f"Discord Bot Token {get_status_str(self.validation_results.get('discord_token', False))}") console.print(f"Discord Bot Token {get_status_str(self.validation_results.get('discord_token', False), self.validation_results.get('discord_bot_name'))}")
console.print(f"Fluxer Bot Token {get_status_str(self.validation_results.get('fluxer_token', False))}") console.print(f"Fluxer Bot Token {get_status_str(self.validation_results.get('fluxer_token', False), self.validation_results.get('fluxer_bot_name'))}")
console.print(f"Discord Server ID {get_status_str(self.validation_results.get('discord_server', False))}") 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))}") 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("Edit now?"):
return return