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

fix configuration container

Browse source
This commit is contained in:
rambros 2026-03-03 02:50:29 +05:30
parent add35ccd1f
commit 1cf5463da6
8 changed files with 937 additions and 1069 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
.gitignore vendored
View file

@ -41,4 +41,9 @@ test_*.py
test_release.zip
test_release/
REAPER-*/
Reaper-*/
DISCORD-*/
FLUXER-*/
REAPER-*/
EXPORT-*/

2
src/core/configuration.py
View file

@ -4,7 +4,7 @@ from pathlib import Path
from pydantic import BaseModel, Field
class MigrationSettings(BaseModel):
batch_size: int = Field(default=50)
batch_size: int = Field(default=100)
rate_limit_delay_seconds: int = Field(default=2)
log_level: str = Field(default="INFO")

297
src/ui/backup_ops.py Normal file
View file

@ -0,0 +1,297 @@
"""
BackupPane – self-contained backup-operations widget.
Embedded inside ModeScreen's "Backup" tab.
"""
import asyncio
import discord
import json
import re
from pathlib import Path
from datetime import datetime
from textual.app import ComposeResult
from textual.containers import Container, Vertical, VerticalScroll
from textual.widgets import Button, Label, Rule
from textual import work
from src.core.configuration import load_config
from src.core.base import MigrationContext
from src.disco_reaper.exporter import DiscordExporter
from src.ui.modals import ProgressModal, ChannelSelectModal
class BackupPane(Container):
"""Backup operations pane — profile, messages, sync."""
DEFAULT_CSS = """
BackupPane { height: auto; width: 100%; }
BackupPane #bp_info {
height: auto; border: tall cyan; padding: 1; margin-bottom: 1;
}
BackupPane #bp_actions { height: auto; }
BackupPane #bp_actions Button { width: 100%; margin-bottom: 1; }
"""
def __init__(self, cfg_name: str, cfg_path: Path, *args, **kwargs):
super().__init__(*args, **kwargs)
self.cfg_name = cfg_name
self.config_path = cfg_path
self.config = load_config(cfg_path)
self.engine = MigrationContext(self.config, target_platform=self.config.target_platform or "fluxer")
self.exporter = DiscordExporter(self.engine.discord_reader, base_dir=f"Reaper-{self.cfg_name}")
def compose(self) -> ComposeResult:
with VerticalScroll():
with Vertical(id="bp_info"):
yield Label("Loading...", id="bp_lbl_server")
yield Label("", id="bp_lbl_bot")
yield Label("", id="bp_lbl_backup")
with Vertical(id="bp_actions"):
yield Button("Backup Server Profile", id="bp_backup_profile", disabled=True)
yield Button("Backup Channel Messages", id="bp_backup_msgs", disabled=True)
yield Button("Update & Sync Backup", id="bp_backup_sync", disabled=True)
def on_mount(self) -> None:
self._validate()
def reload_config(self) -> None:
self.config = load_config(self.config_path)
self.engine = MigrationContext(self.config, target_platform=self.config.target_platform or "fluxer")
self.exporter = DiscordExporter(self.engine.discord_reader, base_dir=f"Reaper-{self.cfg_name}")
self._validate()
# ── validation ────────────────────────────────────────────────────────
@work(exclusive=True, thread=True)
async def _validate(self) -> None:
fillers = ["DISCORD_BOT_TOKEN", "000000000000000000", "DISCORD_SERVER_ID", "", None]
d_token = self.config.discord_bot_token
if d_token in fillers or self.config.discord_server_id in fillers:
self.app.call_from_thread(self._update_ui, "[red]NOT CONFIGURED[/red]", "", "", False)
return
try:
res = await self.engine.discord_reader.validate()
valid = res.get("token", False) and res.get("server", False)
server_name = res.get("server_name", "Unknown")
bot_name = res.get("bot_name", "Unknown")
s_text = f'[green]"{server_name}"[/green]' if valid else "[red]INVALID[/red]"
b_text = f"[green]{bot_name}[/green]" if valid else "[red]INVALID[/red]"
backup_text = ""
info = self._get_backup_info()
if info:
backup_text = f"Last backup: [cyan]{info}[/cyan]"
self.app.call_from_thread(self._update_ui, s_text, b_text, backup_text, valid)
except Exception as e:
self.app.call_from_thread(self._update_ui, f"[red]Error: {e}[/red]", "", "", False)
def _get_backup_info(self) -> str | None:
profile_file = Path(f"Reaper-{self.cfg_name}") / "server_profile.json"
if profile_file.exists():
try:
with open(profile_file, "r", encoding="utf-8") as f:
data = json.load(f)
ts_str = data.get("last_backup")
if ts_str:
dt = datetime.fromisoformat(ts_str)
return dt.strftime("%d-%b-%Y %H:%M")
except Exception:
pass
return None
def _update_ui(self, server_text, bot_text, backup_text, enabled):
self.query_one("#bp_lbl_server", Label).update(f"Source Server: {server_text}")
self.query_one("#bp_lbl_bot", Label).update(f"Bot: {bot_text}")
self.query_one("#bp_lbl_backup", Label).update(backup_text)
for bid in ("#bp_backup_profile", "#bp_backup_msgs", "#bp_backup_sync"):
self.query_one(bid, Button).disabled = not enabled
# ── button routing ────────────────────────────────────────────────────
def on_button_pressed(self, event: Button.Pressed) -> None:
bid = event.button.id
if bid == "bp_backup_profile":
self.run_backup_profile()
elif bid == "bp_backup_msgs":
self.run_backup_messages()
elif bid == "bp_backup_sync":
self.run_backup_sync()
# ── workers ───────────────────────────────────────────────────────────
@work(exclusive=True, thread=True)
async def run_backup_profile(self) -> None:
modal = ProgressModal()
self.app.call_from_thread(self.app.push_screen, modal)
await asyncio.sleep(0.1)
try:
self.app.call_from_thread(modal.set_status, "Starting readers...")
await self.engine.discord_reader.start()
await self.exporter.setup()
self.app.call_from_thread(modal.write, "[yellow]Backing up server profile & skeleton...[/yellow]")
await self.exporter.export_metadata()
await self.exporter.download_server_assets()
self.app.call_from_thread(modal.write, "Exporting structure...")
_, cat_count, chan_count = await self.exporter.export_channels_structure()
self.app.call_from_thread(modal.write, "Exporting assets...")
e_count, s_count = await self.exporter.export_assets()
self.app.call_from_thread(modal.write, f"[bold green]Server Profile backed up to: {self.exporter.export_path}[/bold green]")
self.app.call_from_thread(modal.write, f"- {cat_count} categories & {chan_count} channels")
self.app.call_from_thread(modal.write, f"- {e_count} emojis, {s_count} stickers.")
except discord.Forbidden as e:
self.app.call_from_thread(modal.write, f"[bold red]Backup failed: {e}[/bold red]")
except Exception as e:
self.app.call_from_thread(modal.write, f"[bold red]Error: {e}[/bold red]")
finally:
await self.engine.close_connections()
self.app.call_from_thread(modal.set_status, "Finished.")
self.app.call_from_thread(modal.allow_close)
@work(exclusive=True, thread=True)
async def run_backup_messages(self) -> None:
modal_prog = ProgressModal()
self.app.call_from_thread(self.app.push_screen, modal_prog)
await asyncio.sleep(0.1)
try:
self.app.call_from_thread(modal_prog.set_status, "Fetching channels...")
await self.engine.discord_reader.start()
await self.exporter.setup()
await self.exporter.export_channels_structure()
all_channels = await self.engine.discord_reader.get_channels()
all_categories = await self.engine.discord_reader.get_categories()
cat_map = {c.id: c.name for c in all_categories}
eligible_channels = [
c for c in all_channels
if c.type in [discord.ChannelType.text, discord.ChannelType.news, discord.ChannelType.forum]
]
if not eligible_channels:
self.app.call_from_thread(modal_prog.write, "[yellow]No text/news channels found to backup.[/yellow]")
self.app.call_from_thread(modal_prog.allow_close)
return
any_found = False
backed_up_ids = set()
for chan in eligible_channels:
if (self.exporter.export_path / "message_backup" / f"{chan.id}.json").exists():
any_found = True
backed_up_ids.add(chan.id)
self.app.call_from_thread(self.app.pop_screen)
loop = asyncio.get_running_loop()
future = loop.create_future()
def check_channels(reply: dict | None) -> None:
if not future.done():
future.set_result(reply)
self.app.call_from_thread(
self.app.push_screen,
ChannelSelectModal(eligible_channels, cat_map, backed_up_ids, any_found),
check_channels,
)
reply = await future
if not reply:
return
selected_ids = reply["channels"]
force_overwrite = reply["force"]
selected_channels = [c for c in eligible_channels if c.id in selected_ids]
self.app.call_from_thread(self.app.push_screen, modal_prog)
await asyncio.sleep(0.1)
total_chans = len(selected_channels)
self.app.call_from_thread(modal_prog.set_status, "Backing up messages...")
self.app.call_from_thread(modal_prog.write, f"[yellow]Starting backup for {total_chans} channels...[/yellow]")
for chan in selected_channels:
backup_exists = (self.exporter.export_path / "message_backup" / f"{chan.id}.json").exists()
is_sync = backup_exists and not force_overwrite
label = "Syncing Backup" if is_sync else "Backing up"
self.app.call_from_thread(modal_prog.write, f"[cyan]{label}: {chan.name}[/cyan]")
async def update_msg_count(name, count):
self.app.call_from_thread(modal_prog.set_status, f"{name}: {count} messages")
await self.exporter.export_channel_messages(chan.id, progress_callback=update_msg_count, force=force_overwrite)
await self.exporter.export_threads(chan.id, progress_callback=update_msg_count, force=force_overwrite)
self.app.call_from_thread(modal_prog.write, f"[green]Completed: {chan.name}[/green]")
await self.exporter.export_metadata()
self.app.call_from_thread(modal_prog.write, "[bold green]Message backup complete![/bold green]")
except Exception as e:
self.app.call_from_thread(modal_prog.write, f"[bold red]Message backup failed: {e}[/bold red]")
finally:
await self.engine.close_connections()
self.app.call_from_thread(modal_prog.set_status, "Finished.")
self.app.call_from_thread(modal_prog.allow_close)
@work(exclusive=True, thread=True)
async def run_backup_sync(self) -> None:
modal_prog = ProgressModal()
self.app.call_from_thread(self.app.push_screen, modal_prog)
await asyncio.sleep(0.1)
try:
self.app.call_from_thread(modal_prog.set_status, "Starting sync...")
await self.engine.discord_reader.start()
await self.exporter.setup()
self.app.call_from_thread(modal_prog.write, "Updating structure...")
await self.exporter.export_metadata()
await self.exporter.download_server_assets()
await self.exporter.export_channels_structure()
await self.exporter.export_assets()
all_channels = await self.engine.discord_reader.get_channels()
eligible_channels = [
c for c in all_channels
if c.type in [discord.ChannelType.text, discord.ChannelType.news, discord.ChannelType.forum]
]
selected_channels = [
c for c in eligible_channels
if (self.exporter.export_path / "message_backup" / f"{c.id}.json").exists()
]
if not selected_channels:
self.app.call_from_thread(modal_prog.write, "[yellow]No existing backups found to sync.[/yellow]")
else:
self.app.call_from_thread(modal_prog.write, f"[yellow]Syncing {len(selected_channels)} channels...[/yellow]")
for chan in selected_channels:
self.app.call_from_thread(modal_prog.write, f"[cyan]Syncing: {chan.name}[/cyan]")
async def update_msg_count(name, count):
self.app.call_from_thread(modal_prog.set_status, f"{name}: {count} messages")
await self.exporter.export_channel_messages(chan.id, progress_callback=update_msg_count, force=False)
await self.exporter.export_threads(chan.id, progress_callback=update_msg_count, force=False)
self.app.call_from_thread(modal_prog.write, f"[green]Synced: {chan.name}[/green]")
await self.exporter.export_metadata()
self.app.call_from_thread(modal_prog.write, "[bold green]Sync operation complete![/bold green]")
except Exception as e:
self.app.call_from_thread(modal_prog.write, f"[bold red]Sync failed: {e}[/bold red]")
finally:
await self.engine.close_connections()
self.app.call_from_thread(modal_prog.set_status, "Finished.")
self.app.call_from_thread(modal_prog.allow_close)

567
src/ui/backup_screen.py
View file

@ -1,567 +0,0 @@
import sys
import asyncio
import discord
import logging
import json
import re
from pathlib import Path
from datetime import datetime
from textual.app import App, ComposeResult
from textual.containers import Container, Horizontal, Vertical, VerticalScroll
from textual.widgets import Header, Footer, Button, Static, Label, Input, Checkbox, RadioButton, ProgressBar, RichLog, Rule
from textual.screen import Screen, ModalScreen
from textual import work
from textual.worker import Worker, WorkerState
from src.core.configuration import load_config, save_config
from src.core.base import MigrationContext
from src.disco_reaper.exporter import DiscordExporter
class ConfigModal(ModalScreen[dict]):
"""Modal for configuring token and server ID."""
def __init__(self, current_token: str, current_server: str):
super().__init__()
self.current_token = current_token
self.current_server = current_server
def compose(self) -> ComposeResult:
with Vertical(id="config_dialog"):
yield Label("Configuration", id="config_title")
yield Label("Discord Bot Token:")
yield Input(value=self.current_token, id="token_input")
yield Label("Discord Server ID:")
yield Input(value=self.current_server, id="server_input")
with Horizontal(id="config_buttons"):
yield Button("Save", variant="success", id="btn_save")
yield Button("Cancel", variant="primary", id="btn_cancel")
def on_button_pressed(self, event: Button.Pressed) -> None:
if event.button.id == "btn_save":
token = self.query_one("#token_input", Input).value
server = self.query_one("#server_input", Input).value
self.dismiss({"token": token, "server": server})
elif event.button.id == "btn_cancel":
self.dismiss(None)
class ChannelSelectModal(ModalScreen[dict]):
"""Modal for selecting channels using a simple checkbox list."""
def __init__(self, channels: list, categories: dict, backed_up_ids: set, any_found: bool):
super().__init__()
# Group channels by category
self.channels_by_category = {}
for c in channels:
cat_id = getattr(c, 'category_id', None)
if cat_id not in self.channels_by_category:
self.channels_by_category[cat_id] = []
self.channels_by_category[cat_id].append(c)
self.categories = categories # id -> name
self.backed_up_ids = backed_up_ids
self.any_found = any_found
def compose(self) -> ComposeResult:
with Vertical(id="channel_dialog"):
yield Label(f"Select Channels to Backup", id="chan_title")
with VerticalScroll(id="channel_list_scroll"):
# Group by category
cat_ids = sorted([k for k in self.channels_by_category.keys() if k is not None],
key=lambda k: self.categories.get(k, ""))
# No category channels
if None in self.channels_by_category:
for c in sorted(self.channels_by_category[None], key=lambda x: x.position if hasattr(x, 'position') else 0):
label = f"{c.name}"
color = "green" if c.id in self.backed_up_ids else "white"
yield RadioButton(f"[{color}]{label}[/]", value=False, id=f"chan_{c.id}")
for cat_id in cat_ids:
cat_name = self.categories.get(cat_id, "Unknown Category")
yield Label(f"[cyan]{cat_name}[/cyan]", classes="category_header")
for c in sorted(self.channels_by_category[cat_id], key=lambda x: x.position if hasattr(x, 'position') else 0):
label = f"{c.name}"
color = "green" if c.id in self.backed_up_ids else "white"
yield RadioButton(f"[{color}]{label}[/]", value=False, id=f"chan_{c.id}")
with Horizontal(id="select_all_buttons"):
yield Button("Select All", id="btn_all")
yield Button("Deselect All", id="btn_none")
with Horizontal(id="confirm_buttons"):
if self.any_found:
yield Label("Existing backups found:", classes="label_warning")
yield Button("Sync", variant="success", id="btn_sync")
yield Button("Force Overwrite", variant="error", id="btn_force")
else:
yield Button("Backup", variant="success", id="btn_backup")
yield Button("Cancel", id="btn_cancel_chan")
def on_button_pressed(self, event: Button.Pressed) -> None:
if event.button.id == "btn_all":
for cb in self.query(RadioButton):
cb.value = True
elif event.button.id == "btn_none":
for cb in self.query(RadioButton):
cb.value = False
elif event.button.id in ["btn_sync", "btn_force", "btn_backup"]:
selected = []
for cb in self.query(RadioButton):
if cb.value and cb.id and cb.id.startswith("chan_"):
chan_id = int(cb.id.split("_")[1])
selected.append(chan_id)
if not selected:
return
force = event.button.id == "btn_force"
self.dismiss({"channels": selected, "force": force})
elif event.button.id == "btn_cancel_chan":
self.dismiss(None)
class ProgressModal(ModalScreen[None]):
"""Modal to show progress of backup."""
def compose(self) -> ComposeResult:
with Vertical(id="progress_dialog"):
yield Label("Operation Status", id="progress_status")
yield ProgressBar(total=None, show_eta=False, id="progress_bar")
yield RichLog(id="progress_log", highlight=True, markup=True)
yield Button("Close", id="btn_close_progress", disabled=True)
def on_button_pressed(self, event: Button.Pressed):
if event.button.id == "btn_close_progress":
self.dismiss(None)
def write_to_log(self, message: str):
self.query_one("#progress_log", RichLog).write(message)
def set_status(self, status: str):
self.query_one("#progress_status", Label).update(status)
def allow_close(self):
btn = self.query_one("#btn_close_progress", Button)
btn.disabled = False
btn.variant = "success"
self.query_one("#progress_bar", ProgressBar).update(total=100, progress=100)
class BackupScreen(Screen):
"""Main screen of the application."""
CSS = """
#backup_scroll {
align: center middle;
}
#main_container {
width: 80%;
height: auto;
min-height: 20;
border: solid green;
padding: 1 2;
margin: 2 0;
}
#info_container {
height: auto;
margin-bottom: 2;
border: tall cyan;
padding: 1;
}
#actions_container {
height: auto;
layout: vertical;
align: center top;
margin-top: 1;
}
Button {
width: 100%;
margin-bottom: 1;
}
#config_dialog {
width: 60%;
height: 60%;
border: thick $background 80%;
background: $surface;
padding: 1 2;
}
#config_title { text-style: bold; margin-bottom: 1; }
#config_buttons { height: auto; margin-top: 1; }
#config_buttons Button { width: 1fr; margin: 0 1; }
#channel_dialog {
width: 80%;
height: 80%;
border: thick $background 80%;
background: $surface;
padding: 1 2;
}
#chan_title { text-style: bold; margin-bottom: 1; }
.category_header {
margin-top: 1;
background: $primary 10%;
text-style: bold;
padding-left: 1;
}
#channel_list_scroll {
height: 1fr;
border: solid $primary;
margin-bottom: 1;
padding: 0 1;
}
#select_all_buttons { height: auto; margin-bottom: 1; }
#select_all_buttons Button { width: auto; margin-right: 1; }
#confirm_buttons { height: auto; margin-top: 1; }
#confirm_buttons Button { width: auto; margin: 0 1; }
RadioButton:focus {
background: transparent;
border: none;
color: $text;
}
RadioButton > .radio-button--label {
padding: 0 1;
}
RadioButton:focus > .radio-button--label {
background: transparent;
text-style: none;
}
#progress_dialog {
width: 80%;
height: 80%;
border: thick $background 80%;
background: $surface;
padding: 1 2;
}
#progress_status { text-style: bold; margin-bottom: 1; }
#progress_bar { margin-bottom: 1; }
#progress_log { height: 1fr; margin-bottom: 1; border: solid $primary; }
.label_warning { color: yellow; margin-right: 1; content-align: center middle;}
"""
BINDINGS = [
("q", "app.exit", "Quit"),
("c", "config", "Config"),
("b", "app.pop_screen", "Back"),
]
def __init__(self, cfg_name: str, cfg_path: Path, *args, **kwargs):
super().__init__(*args, **kwargs)
self.cfg_name = cfg_name
self.config_path = cfg_path
self.config = load_config(self.config_path)
self.engine = MigrationContext(self.config, target_platform="fluxer")
self.exporter = DiscordExporter(self.engine.discord_reader, base_dir=f"Reaper-{self.cfg_name}")
self.validation_results = {}
self.tokens_valid = False
def compose(self) -> ComposeResult:
yield Header(show_clock=True)
with VerticalScroll(id="backup_scroll"):
with Container(id="main_container"):
yield Label("Disco Reaper - Server Backup Tool", id="title_label")
with Vertical(id="info_container"):
yield Label("Loading...", id="lbl_server")
yield Label("", id="lbl_bot")
yield Label("", id="lbl_backup")
with Vertical(id="actions_container"):
yield Button("Backup Server Profile", id="btn_backup_profile", disabled=True)
yield Button("Backup Channel Messages", id="btn_backup_msgs", disabled=True)
yield Button("Update & Sync Backup", id="btn_backup_sync", disabled=True)
yield Rule()
yield Button("Configuration", id="btn_config")
yield Button("Back", id="btn_back")
yield Button("Exit", id="btn_exit", variant="error")
yield Footer()
def on_mount(self) -> None:
self.validate_config()
@work(exclusive=True, thread=True)
async def validate_config(self) -> None:
def update_ui(server_text, bot_text, backup_text, buttons_enabled):
self.query_one("#lbl_server", Label).update(f"Source Server: {server_text}")
self.query_one("#lbl_bot", Label).update(f"Bot name: {bot_text}")
self.query_one("#lbl_backup", Label).update(backup_text)
for btn_id in ["#btn_backup_profile", "#btn_backup_msgs", "#btn_backup_sync"]:
self.query_one(btn_id, Button).disabled = not buttons_enabled
d_token = self.config.discord_bot_token
fillers = ["DISCORD_BOT_TOKEN", "000000000000000000", "DISCORD_SERVER_ID", "", None]
if d_token in fillers or self.config.discord_server_id in fillers:
self.app.call_from_thread(update_ui, "[red]NOT CONNECTED[/red]", "[red]UNKNOWN[/red]", "", False)
self.tokens_valid = False
return
self.app.call_from_thread(update_ui, "[yellow]Validating...[/yellow]", "[yellow]Validating...[/yellow]", "", False)
try:
res = await self.engine.discord_reader.validate()
except Exception as e:
res = {}
self.app.call_from_thread(update_ui, f"[red]Validation Error: {e}[/red]", "", "", False)
self.tokens_valid = False
return
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_name"] = res.get("server_name")
self.tokens_valid = res.get("token") and res.get("server")
d_name = self.validation_results.get("discord_server_name")
server_display = f"[green]\"{d_name}\"[/green]" if d_name else "[red]NOT CONNECTED[/red]"
b_name = self.validation_results.get("discord_bot_name")
bot_display = f"[green]\"{b_name}\"[/green]" if b_name else "[red]UNKNOWN[/red]"
backup_text = ""
if self.tokens_valid:
backup_ts = self.get_backup_info()
if backup_ts:
backup_text = f"Backup Found: [yellow]{backup_ts}[/yellow]"
self.app.call_from_thread(update_ui, server_display, bot_display, backup_text, self.tokens_valid)
def get_backup_info(self):
d_name = self.validation_results.get("discord_server_name")
d_id = self.config.discord_server_id
if not d_name or not d_id:
return None
export_path = Path(f"Reaper-{self.cfg_name}") / f"DISCORD-{d_id}"
profile_file = export_path / "server_profile.json"
if profile_file.exists():
try:
with open(profile_file, "r", encoding="utf-8") as f:
data = json.load(f)
ts_str = data.get("last_backup")
if ts_str:
dt = datetime.fromisoformat(ts_str)
return dt.strftime("%d-%b-%Y %H:%M")
except Exception:
pass
return None
def action_config(self) -> None:
self.open_config()
def open_config(self) -> None:
def check_reply(reply: dict | None) -> None:
if reply is not None:
self.config.discord_bot_token = reply["token"]
self.config.discord_server_id = reply["server"]
save_config(self.config, self.config_path)
self.engine = MigrationContext(self.config, target_platform="fluxer")
self.exporter = DiscordExporter(self.engine.discord_reader, base_dir=f"Reaper-{self.cfg_name}")
self.validate_config()
self.app.push_screen(
ConfigModal(self.config.discord_bot_token, self.config.discord_server_id),
check_reply
)
def on_button_pressed(self, event: Button.Pressed) -> None:
if event.button.id == "btn_exit":
self.app.exit()
elif event.button.id == "btn_back":
self.app.pop_screen()
elif event.button.id == "btn_config":
self.open_config()
elif event.button.id == "btn_backup_profile":
self.run_backup_profile()
elif event.button.id == "btn_backup_msgs":
self.run_backup_messages()
elif event.button.id == "btn_backup_sync":
self.run_backup_sync()
@work(exclusive=True, thread=True)
async def run_backup_profile(self) -> None:
modal = ProgressModal()
self.app.call_from_thread(self.app.push_screen, modal)
await asyncio.sleep(0.1) # Wait for mount
try:
self.app.call_from_thread(modal.set_status, "Starting readers...")
await self.engine.discord_reader.start()
meta = await self.exporter.setup()
self.app.call_from_thread(modal.write_to_log, "[yellow]Backing up server profile & skeleton...[/yellow]")
await self.exporter.export_metadata()
await self.exporter.download_server_assets()
self.app.call_from_thread(modal.write_to_log, "Exporting structure...")
_, cat_count, chan_count = await self.exporter.export_channels_structure()
self.app.call_from_thread(modal.write_to_log, "Exporting assets...")
e_count, s_count = await self.exporter.export_assets()
self.app.call_from_thread(modal.write_to_log, f"[bold green]Server Profile backed up to: {self.exporter.export_path}[/bold green]")
self.app.call_from_thread(modal.write_to_log, f"- {cat_count} categories & {chan_count} channels")
self.app.call_from_thread(modal.write_to_log, f"- {e_count} emojis, {s_count} stickers.")
except discord.Forbidden as e:
self.app.call_from_thread(modal.write_to_log, f"[bold red]Backup failed: {e}[/bold red]")
except Exception as e:
self.app.call_from_thread(modal.write_to_log, f"[bold red]Error: {e}[/bold red]")
finally:
await self.engine.close_connections()
self.app.call_from_thread(modal.set_status, "Finished.")
self.app.call_from_thread(modal.allow_close)
@work(exclusive=True, thread=True)
async def run_backup_messages(self) -> None:
modal_prog = ProgressModal()
self.app.call_from_thread(self.app.push_screen, modal_prog)
await asyncio.sleep(0.1)
try:
self.app.call_from_thread(modal_prog.set_status, "Fetching channels...")
await self.engine.discord_reader.start()
await self.exporter.setup()
await self.exporter.export_channels_structure()
all_channels = await self.engine.discord_reader.get_channels()
all_categories = await self.engine.discord_reader.get_categories()
cat_map = {c.id: c.name for c in all_categories}
eligible_channels = [
c for c in all_channels
if c.type in [discord.ChannelType.text, discord.ChannelType.news, discord.ChannelType.forum]
]
if not eligible_channels:
self.app.call_from_thread(modal_prog.write_to_log, "[yellow]No text/news channels found to backup.[/yellow]")
self.app.call_from_thread(modal_prog.allow_close)
return
any_found = False
backed_up_ids = set()
for chan in eligible_channels:
if (self.exporter.export_path / "message_backup" / f"{chan.id}.json").exists():
any_found = True
backed_up_ids.add(chan.id)
# Need to ask for channels. We temporarily pop the progress modal, push channel select, wait for result, push progress again.
self.app.call_from_thread(self.app.pop_screen)
# Setup a future to await modal response
loop = asyncio.get_running_loop()
future = loop.create_future()
def check_channels(reply: dict | None) -> None:
if not future.done():
future.set_result(reply)
self.app.call_from_thread(
self.app.push_screen,
ChannelSelectModal(eligible_channels, cat_map, backed_up_ids, any_found),
check_channels
)
reply = await future
if not reply:
# Cancelled
return
selected_ids = reply["channels"]
force_overwrite = reply["force"]
selected_channels = [c for c in eligible_channels if c.id in selected_ids]
self.app.call_from_thread(self.app.push_screen, modal_prog)
await asyncio.sleep(0.1)
total_chans = len(selected_channels)
self.app.call_from_thread(modal_prog.set_status, "Backing up messages...")
self.app.call_from_thread(modal_prog.write_to_log, f"[yellow]Starting backup for {total_chans} channels...[/yellow]")
for chan in selected_channels:
backup_exists = (self.exporter.export_path / "message_backup" / f"{chan.id}.json").exists()
is_sync = backup_exists and not force_overwrite
label = "Syncing Backup" if is_sync else "Backing up"
self.app.call_from_thread(modal_prog.write_to_log, f"[cyan]{label}: {chan.name}[/cyan]")
async def update_msg_count(name, count):
self.app.call_from_thread(modal_prog.set_status, f"{name}: {count} messages")
await self.exporter.export_channel_messages(chan.id, progress_callback=update_msg_count, force=force_overwrite)
await self.exporter.export_threads(chan.id, progress_callback=update_msg_count, force=force_overwrite)
self.app.call_from_thread(modal_prog.write_to_log, f"[green]Completed: {chan.name}[/green]")
await self.exporter.export_metadata()
self.app.call_from_thread(modal_prog.write_to_log, "[bold green]Message backup complete![/bold green]")
except Exception as e:
self.app.call_from_thread(modal_prog.write_to_log, f"[bold red]Message backup failed: {e}[/bold red]")
finally:
await self.engine.close_connections()
self.app.call_from_thread(modal_prog.set_status, "Finished.")
self.app.call_from_thread(modal_prog.allow_close)
@work(exclusive=True, thread=True)
async def run_backup_sync(self) -> None:
modal_prog = ProgressModal()
self.app.call_from_thread(self.app.push_screen, modal_prog)
await asyncio.sleep(0.1)
try:
self.app.call_from_thread(modal_prog.set_status, "Starting sync...")
await self.engine.discord_reader.start()
await self.exporter.setup()
self.app.call_from_thread(modal_prog.write_to_log, "Updating structure...")
await self.exporter.export_metadata()
await self.exporter.download_server_assets()
await self.exporter.export_channels_structure()
await self.exporter.export_assets()
all_channels = await self.engine.discord_reader.get_channels()
eligible_channels = [
c for c in all_channels
if c.type in [discord.ChannelType.text, discord.ChannelType.news, discord.ChannelType.forum]
]
selected_channels = [
c for c in eligible_channels
if (self.exporter.export_path / "message_backup" / f"{c.id}.json").exists()
]
if not selected_channels:
self.app.call_from_thread(modal_prog.write_to_log, "[yellow]No existing backups found to sync.[/yellow]")
else:
self.app.call_from_thread(modal_prog.write_to_log, f"[yellow]Syncing {len(selected_channels)} channels...[/yellow]")
for chan in selected_channels:
self.app.call_from_thread(modal_prog.write_to_log, f"[cyan]Syncing: {chan.name}[/cyan]")
async def update_msg_count(name, count):
self.app.call_from_thread(modal_prog.set_status, f"{name}: {count} messages")
await self.exporter.export_channel_messages(chan.id, progress_callback=update_msg_count, force=False)
await self.exporter.export_threads(chan.id, progress_callback=update_msg_count, force=False)
self.app.call_from_thread(modal_prog.write_to_log, f"[green]Synced: {chan.name}[/green]")
await self.exporter.export_metadata()
self.app.call_from_thread(modal_prog.write_to_log, "[bold green]Sync operation complete![/bold green]")
except Exception as e:
self.app.call_from_thread(modal_prog.write_to_log, f"[bold red]Sync failed: {e}[/bold red]")
finally:
await self.engine.close_connections()
self.app.call_from_thread(modal_prog.set_status, "Finished.")
self.app.call_from_thread(modal_prog.allow_close)

43
src/ui/main_app.py
View file

@ -114,7 +114,8 @@ class ConfigSelectionScreen(Screen):
def on_list_view_selected(self, event: ListView.Selected) -> None:
cfg_name = event.item.name
cfg_path = Path(f"Reaper-{cfg_name}") / "config.yaml"
self.app.push_screen(ConfigScreen(cfg_name, cfg_path))
from src.ui.mode_screen import ModeScreen
self.app.push_screen(ModeScreen(cfg_name, cfg_path))
def on_button_pressed(self, event: Button.Pressed) -> None:
if event.button.id == "btn_new_config":
@ -157,11 +158,12 @@ class ConfigScreen(Screen):
DEFAULT_CSS = """
ConfigScreen { align: center middle; }
#cfg_scroll { width: 100%; height: 100%; align: center middle; }
#cfg_outer { width: 100%; height: 100%; align: center top; }
#cfg_container {
width: 70; height: auto;
border: solid green; padding: 1 2; margin: 1 0;
width: 80%; height: 100%; layout: vertical;
border: solid green; padding: 1 2; margin: 2 0;
}
#cfg_scroll { width: 100%; height: 1fr; margin-bottom: 1; }
#cfg_title {
text-style: bold; color: green; margin-bottom: 1;
content-align: center middle; width: 100%;
@ -175,7 +177,7 @@ class ConfigScreen(Screen):
height: auto; margin: 0 0 0 2;
}
#target_section { height: auto; }
#cfg_actions { height: auto; margin-top: 1; }
#cfg_actions { height: auto; margin-top: 1; margin-bottom: 0; dock: bottom; }
#cfg_actions Button { width: 1fr; margin: 0 1; }
"""
@ -189,13 +191,13 @@ class ConfigScreen(Screen):
def compose(self) -> ComposeResult:
yield Header(show_clock=True)
with VerticalScroll(id="cfg_scroll"):
with Container(id="cfg_outer"):
with Container(id="cfg_container"):
yield Label(f"Configuration — {self.cfg_name}", id="cfg_title")
with VerticalScroll(id="cfg_scroll"):
# ── Discord ──────────────────────────────────────────────
yield Label("Discord", classes="section_title")
yield Rule()
yield Label("Bot Token:", classes="field_label")
yield Input(
value=self.config.discord_bot_token or "",
@ -210,7 +212,6 @@ class ConfigScreen(Screen):
# ── Reaper Mode ──────────────────────────────────────────
yield Label("Reaper Mode", classes="section_title")
yield Rule()
cur_mode = self.config.tool_mode or "backup_only"
with RadioSet(id="mode_radio"):
yield RadioButton(
@ -232,7 +233,6 @@ class ConfigScreen(Screen):
# ── Target Platform (hidden for backup_only) ─────────────
with Vertical(id="target_section"):
yield Label("Target Platform", classes="section_title")
yield Rule()
cur_plat = self.config.target_platform or "none"
with RadioSet(id="plat_radio"):
yield RadioButton(
@ -256,11 +256,15 @@ class ConfigScreen(Screen):
value=self.config.target_server_id or "",
id="inp_target_server",
)
yield Label("Target API URL:", classes="field_label")
yield Input(
value=self.config.target_api_url or "default",
id="inp_target_api",
)
yield Rule()
with Horizontal(id="cfg_actions"):
yield Button("Save & Start", variant="success", id="btn_start")
yield Button("Save", variant="primary", id="btn_save")
yield Button("Save Configuration", variant="primary", id="btn_save")
yield Button("Back", id="btn_back")
yield Footer()
@ -300,22 +304,14 @@ class ConfigScreen(Screen):
self.config.target_platform = self._get_selected_platform()
self.config.target_bot_token = self.query_one("#inp_target_token", Input).value.strip() or self.config.target_bot_token
self.config.target_server_id = self.query_one("#inp_target_server", Input).value.strip() or self.config.target_server_id
self.config.target_api_url = self.query_one("#inp_target_api", Input).value.strip() or self.config.target_api_url
else:
self.config.target_platform = "none"
save_config(self.config, self.cfg_path)
def _launch_mode(self) -> None:
mode = self.config.tool_mode
if mode == "direct_transfer":
from src.ui.shuttle_screen import ShuttleScreen
self.app.push_screen(ShuttleScreen(self.cfg_name, self.cfg_path))
elif mode == "backup_transfer":
from src.ui.backup_screen import BackupScreen
self.app.push_screen(BackupScreen(self.cfg_name, self.cfg_path))
else: # backup_only
from src.ui.backup_screen import BackupScreen
self.app.push_screen(BackupScreen(self.cfg_name, self.cfg_path))
pass # No longer needed
def action_go_back(self) -> None:
self.app.pop_screen()
@ -326,9 +322,7 @@ class ConfigScreen(Screen):
elif event.button.id == "btn_save":
self._collect_and_save()
self.notify("Configuration saved.", severity="information")
elif event.button.id == "btn_start":
self._collect_and_save()
self._launch_mode()
self.dismiss(True)
# ──────────────────────────────────────────────────────────────────────────────
@ -336,6 +330,7 @@ class ConfigScreen(Screen):
# ──────────────────────────────────────────────────────────────────────────────
class ReaperApp(App):
theme = "dracula"
def on_mount(self) -> None:
self.push_screen(ConfigSelectionScreen())

228
src/ui/modals.py Normal file
View file

@ -0,0 +1,228 @@
"""
Shared modals used by backup and shuttle operations.
"""
from textual.app import ComposeResult
from textual.containers import Horizontal, Vertical, VerticalScroll
from textual.widgets import Button, Label, Input, ProgressBar, RichLog, Rule, RadioButton
from textual.screen import ModalScreen
# ---------------------------------------------------------------------------
# ProgressModal – unified progress / log display
# ---------------------------------------------------------------------------
class ProgressModal(ModalScreen[None]):
"""Modal to display progress for any long-running operation."""
def compose(self) -> ComposeResult:
with Vertical(id="progress_dialog"):
yield Label("Operation Status", id="progress_status")
yield ProgressBar(total=None, show_eta=False, id="progress_bar")
yield RichLog(id="progress_log", highlight=True, markup=True)
yield Button("Close", id="btn_close_progress", disabled=True)
def on_button_pressed(self, event: Button.Pressed):
if event.button.id == "btn_close_progress":
self.dismiss(None)
def write(self, message: str):
self.query_one("#progress_log", RichLog).write(message)
# alias used by backup code
write_to_log = write
def set_status(self, status: str):
self.query_one("#progress_status", Label).update(status)
def set_progress(self, current: int, total: int):
bar = self.query_one("#progress_bar", ProgressBar)
bar.update(total=total, progress=current)
def allow_close(self):
btn = self.query_one("#btn_close_progress", Button)
btn.disabled = False
btn.variant = "success"
self.query_one("#progress_bar", ProgressBar).update(total=100, progress=100)
# ---------------------------------------------------------------------------
# ConfirmModal – simple yes / no
# ---------------------------------------------------------------------------
class ConfirmModal(ModalScreen[bool]):
"""Simple Yes / No confirmation modal."""
def __init__(self, message: str, danger: bool = False):
super().__init__()
self._message = message
self._danger = danger
def compose(self) -> ComposeResult:
with Vertical(id="confirm_dialog"):
yield Label(self._message, id="confirm_msg")
with Horizontal(id="confirm_buttons"):
yield Button("Confirm", variant="error" if self._danger else "success", id="btn_yes")
yield Button("Cancel", variant="primary", id="btn_no")
def on_button_pressed(self, event: Button.Pressed):
self.dismiss(event.button.id == "btn_yes")
# ---------------------------------------------------------------------------
# SubMenuModal – generic labelled-button list
# ---------------------------------------------------------------------------
class SubMenuModal(ModalScreen[str]):
"""A generic sub-menu modal that presents a list of labelled buttons."""
def __init__(self, title: str, options: list[tuple[str, str, str]]):
"""options: list of (button_id, label, variant)"""
super().__init__()
self._title = title
self._options = options
def compose(self) -> ComposeResult:
with Vertical(id="submenu_dialog"):
yield Label(self._title, id="submenu_title")
for btn_id, label, variant in self._options:
yield Button(label, id=btn_id, variant=variant)
yield Rule()
yield Button("Cancel", id="btn_cancel_sub")
def on_button_pressed(self, event: Button.Pressed):
if event.button.id == "btn_cancel_sub":
self.dismiss(None)
else:
self.dismiss(event.button.id)
# ---------------------------------------------------------------------------
# ChannelPickerModal – single-channel selection (shuttle)
# ---------------------------------------------------------------------------
class ChannelPickerModal(ModalScreen[int]):
"""Modal listing Discord channels for single-channel selection."""
def __init__(self, channels: list, categories: dict, label: str = "Select Channel"):
super().__init__()
self._channels = channels
self._categories = categories
self._label = label
def compose(self) -> ComposeResult:
with Vertical(id="chanpick_dialog"):
yield Label(self._label, id="chanpick_title")
with VerticalScroll(id="chanpick_scroll"):
cat_grouped: dict[int | None, list] = {}
for c in self._channels:
cat_id = getattr(c, "category_id", None) if not isinstance(c, dict) else c.get("parent_id")
cat_grouped.setdefault(cat_id, []).append(c)
for cat_id in sorted(cat_grouped, key=lambda k: self._categories.get(k, "") if k else ""):
if cat_id is not None and cat_id in self._categories:
yield Label(f"[cyan]{self._categories[cat_id]}[/cyan]", classes="category_header")
for c in cat_grouped[cat_id]:
if isinstance(c, dict):
name = c.get("name", "Unnamed")
cid = c.get("id")
else:
name = c.name
cid = c.id
yield RadioButton(name, value=False, id=f"chpk_{cid}")
with Horizontal(id="chanpick_buttons"):
yield Button("Select", variant="success", id="btn_pick_ok")
yield Button("Cancel", id="btn_pick_cancel")
def on_button_pressed(self, event: Button.Pressed):
if event.button.id == "btn_pick_cancel":
self.dismiss(None)
elif event.button.id == "btn_pick_ok":
for rb in self.query(RadioButton):
if rb.value and rb.id and rb.id.startswith("chpk_"):
self.dismiss(int(rb.id.split("_", 1)[1]))
return
# Nothing selected
return
# ---------------------------------------------------------------------------
# ChannelSelectModal – multi-channel selection with sync/force (backup)
# ---------------------------------------------------------------------------
class ChannelSelectModal(ModalScreen[dict]):
"""Modal for selecting channels using a simple checkbox list."""
def __init__(self, channels: list, categories: dict, backed_up_ids: set, any_found: bool):
super().__init__()
# Group channels by category
self.channels_by_category = {}
for c in channels:
cat_id = getattr(c, 'category_id', None)
if cat_id not in self.channels_by_category:
self.channels_by_category[cat_id] = []
self.channels_by_category[cat_id].append(c)
self.categories = categories # id -> name
self.backed_up_ids = backed_up_ids
self.any_found = any_found
def compose(self) -> ComposeResult:
with Vertical(id="channel_dialog"):
yield Label("Select Channels to Backup", id="chan_title")
with VerticalScroll(id="channel_list_scroll"):
cat_ids = sorted(
[k for k in self.channels_by_category.keys() if k is not None],
key=lambda k: self.categories.get(k, ""),
)
# No category channels
if None in self.channels_by_category:
for c in sorted(self.channels_by_category[None], key=lambda x: x.position if hasattr(x, 'position') else 0):
label = f"{c.name}"
color = "green" if c.id in self.backed_up_ids else "white"
yield RadioButton(f"[{color}]{label}[/]", value=False, id=f"chan_{c.id}")
for cat_id in cat_ids:
cat_name = self.categories.get(cat_id, "Unknown Category")
yield Label(f"[cyan]{cat_name}[/cyan]", classes="category_header")
for c in sorted(self.channels_by_category[cat_id], key=lambda x: x.position if hasattr(x, 'position') else 0):
label = f"{c.name}"
color = "green" if c.id in self.backed_up_ids else "white"
yield RadioButton(f"[{color}]{label}[/]", value=False, id=f"chan_{c.id}")
with Horizontal(id="select_all_buttons"):
yield Button("Select All", id="btn_all")
yield Button("Deselect All", id="btn_none")
with Horizontal(id="confirm_buttons"):
if self.any_found:
yield Label("Existing backups found:", classes="label_warning")
yield Button("Sync", variant="success", id="btn_sync")
yield Button("Force Overwrite", variant="error", id="btn_force")
else:
yield Button("Backup", variant="success", id="btn_backup")
yield Button("Cancel", id="btn_cancel_chan")
def on_button_pressed(self, event: Button.Pressed) -> None:
if event.button.id == "btn_all":
for cb in self.query(RadioButton):
cb.value = True
elif event.button.id == "btn_none":
for cb in self.query(RadioButton):
cb.value = False
elif event.button.id in ["btn_sync", "btn_force", "btn_backup"]:
selected = []
for cb in self.query(RadioButton):
if cb.value and cb.id and cb.id.startswith("chan_"):
chan_id = int(cb.id.split("_")[1])
selected.append(chan_id)
if not selected:
return
force = event.button.id == "btn_force"
self.dismiss({"channels": selected, "force": force})
elif event.button.id == "btn_cancel_chan":
self.dismiss(None)

202
src/ui/mode_screen.py Normal file
View file

@ -0,0 +1,202 @@
"""
ModeScreen – the single unified screen used for all tool_mode values.
Shows the appropriate pane(s) based on the mode:
- backup_only: Backup pane only
- direct_transfer: Migrate pane only
- backup_transfer: Both panes with a toggle button to switch between them
"""
from pathlib import Path
from textual.app import ComposeResult
from textual.containers import Container, Horizontal, VerticalScroll
from textual.widgets import Header, Footer, Button, ContentSwitcher, Rule
from textual.screen import Screen
from src.core.configuration import load_config
from src.ui.backup_ops import BackupPane
from src.ui.shuttle_ops import ShuttlePane
class ModeScreen(Screen):
"""Unified mode screen — adapts content based on tool_mode."""
CSS = """
#main_scroll {
align: center top;
height: 100%;
}
#main_container {
width: 80%;
height: 100%;
layout: vertical;
min-height: 20;
border: solid green;
padding: 1 2;
margin: 2 0;
}
#switcher {
height: 1fr;
}
#bottom_actions {
height: auto;
dock: bottom;
margin: 1 1 0 1;
}
#bottom_actions Button, #btn_switch {
width: 1fr;
margin: 0 1;
}
/* Modal styling (shared across both panes) */
#progress_dialog {
width: 80%;
height: 80%;
border: thick $background 80%;
background: $surface;
padding: 1 2;
}
#progress_status { text-style: bold; margin-bottom: 1; }
#progress_bar { margin-bottom: 1; }
#progress_log { height: 1fr; margin-bottom: 1; border: solid $primary; }
#shuttle_config_dialog, #platform_select_dialog, #submenu_dialog, #confirm_dialog {
width: 60%;
height: auto;
max-height: 80%;
border: thick $background 80%;
background: $surface;
padding: 1 2;
}
#chanpick_dialog {
width: 70%;
height: 75%;
border: thick $background 80%;
background: $surface;
padding: 1 2;
}
#chanpick_scroll {
height: 1fr;
border: solid $primary;
margin-bottom: 1;
padding: 0 1;
}
.category_header {
margin-top: 1;
background: $primary 10%;
text-style: bold;
padding-left: 1;
}
#config_title, #platform_title, #submenu_title, #confirm_msg, #chanpick_title, #chan_title {
text-style: bold; margin-bottom: 1;
}
#config_buttons, #confirm_buttons, #chanpick_buttons {
height: auto; margin-top: 1;
}
#config_buttons Button, #confirm_buttons Button, #chanpick_buttons Button {
width: 1fr; margin: 0 1;
}
#channel_dialog {
width: 80%;
height: 80%;
border: thick $background 80%;
background: $surface;
padding: 1 2;
}
#channel_list_scroll {
height: 1fr;
border: solid $primary;
margin-bottom: 1;
padding: 0 1;
}
#select_all_buttons { height: auto; margin-bottom: 1; }
#select_all_buttons Button { width: auto; margin-right: 1; }
RadioButton:focus { background: transparent; border: none; }
RadioButton > .radio-button--label { padding: 0 1; }
RadioButton:focus > .radio-button--label { background: transparent; text-style: none; }
"""
BINDINGS = [
("q", "app.exit", "Quit"),
]
def __init__(self, cfg_name: str, cfg_path: Path, *args, **kwargs):
super().__init__(*args, **kwargs)
self.cfg_name = cfg_name
self.config_path = cfg_path
self.config = load_config(cfg_path)
self._showing_backup = True # track which pane is visible in combined mode
def compose(self) -> ComposeResult:
yield Header(show_clock=True)
mode = self.config.tool_mode or "backup_only"
with VerticalScroll(id="main_scroll"):
with Container(id="main_container"):
if mode == "backup_only":
yield BackupPane(self.cfg_name, self.config_path, id="pane_backup")
elif mode == "direct_transfer":
yield ShuttlePane(self.cfg_name, self.config_path, id="pane_migrate")
else: # backup_transfer
with ContentSwitcher(initial="pane_backup", id="switcher"):
yield BackupPane(self.cfg_name, self.config_path, id="pane_backup")
yield ShuttlePane(self.cfg_name, self.config_path, id="pane_migrate")
yield Rule()
with Horizontal(id="bottom_actions"):
if mode == "backup_transfer":
yield Button("Switch to Migrate ⇄", id="btn_switch", variant="primary")
yield Button("Configuration", id="btn_config")
yield Button("Exit", id="btn_exit", variant="error")
yield Footer()
def on_button_pressed(self, event: Button.Pressed) -> None:
bid = event.button.id
if bid == "btn_exit":
self.app.exit()
elif bid == "btn_config":
from src.ui.main_app import ConfigScreen
def reload_screen(saved: bool = False):
if saved:
self.app.pop_screen()
from src.ui.mode_screen import ModeScreen
self.app.push_screen(ModeScreen(self.cfg_name, self.config_path))
self.app.push_screen(ConfigScreen(self.cfg_name, self.config_path), reload_screen)
elif bid == "btn_switch":
self._toggle_pane()
def _toggle_pane(self) -> None:
"""Switch between Backup and Migrate panes in backup_transfer mode."""
switcher = self.query_one("#switcher", ContentSwitcher)
btn = self.query_one("#btn_switch", Button)
if self._showing_backup:
switcher.current = "pane_migrate"
btn.label = "Switch to Backup ⇄"
self._showing_backup = False
else:
switcher.current = "pane_backup"
btn.label = "Switch to Migrate ⇄"
self._showing_backup = True
def on_screen_resume(self) -> None:
"""Reload config when returning from ConfigScreen.
If the mode changed, pop back to ConfigSelectionScreen so the user
re-enters with the correct layout.
"""
old_mode = self.config.tool_mode
self.config = load_config(self.config_path)
new_mode = self.config.tool_mode
if new_mode != old_mode:
self.app.pop_screen()
return
# Propagate config reload to panes
for pane in self.query(BackupPane):
pane.reload_config()
for pane in self.query(ShuttlePane):
pane.reload_config()

544
src/ui/shuttle_screen.py → src/ui/shuttle_ops.py
View file

@ -1,10 +1,8 @@
"""
Shuttle Screen – native Textual TUI for direct server-to-server migration.
Ports every operation from the old Rich CLI (MigrationCLI) into Textual
Screens, Modals, and Workers.
ShuttlePane – self-contained shuttle (migration) operations widget.
Embedded inside ModeScreen's "Migrate" tab.
"""
import sys
import asyncio
import discord
import logging
@ -13,20 +11,17 @@ import time
import aiohttp
from pathlib import Path
from textual.app import App, ComposeResult
from textual.containers import Container, Horizontal, Vertical, VerticalScroll
from textual.widgets import (
Header, Footer, Button, Static, Label, Input,
Checkbox, RadioButton, ProgressBar, RichLog, Rule,
ListItem, ListView, RadioSet,
)
from textual.screen import Screen, ModalScreen
from textual.app import ComposeResult
from textual.containers import Container, Vertical, VerticalScroll
from textual.widgets import Button, Label, Rule
from textual import work
from textual.worker import Worker, WorkerState
from src.core.configuration import load_config, save_config
from src.core.configuration import load_config
from src.core.base import MigrationContext
from src.core.audit import log_audit_event
from src.ui.modals import (
ProgressModal, ConfirmModal, SubMenuModal, ChannelPickerModal,
)
import src.fluxer.roles_permissions as fluxer_roles
import src.stoat.roles_permissions as stoat_roles
@ -52,329 +47,81 @@ class RateLimitHandler(logging.Handler):
super().__init__()
def emit(self, record):
try:
msg = record.getMessage()
if "retry" in msg.lower() and ("rate limit" in msg.lower() or "429" in msg):
match = re.search(r"in ([\d.]+)\s*(?:seconds?|s)", msg, re.IGNORECASE)
if match:
seconds = match.group(1)
platform = "API"
if "discord" in record.name.lower():
platform = "Discord"
elif "fluxer" in record.name.lower():
platform = "Fluxer"
elif "stoat" in record.name.lower():
platform = "Stoat"
global global_rate_limit_msg, global_rate_limit_expires
global_rate_limit_msg = f"{platform} rate limit {seconds}s"
msg = record.getMessage()
if "rate" in msg.lower() and "limit" in msg.lower():
global_rate_limit_msg = msg
try:
global_rate_limit_expires = time.time() + float(seconds)
parts = msg.split()
for i, p in enumerate(parts):
if p.lower() in ("retry_after", "retry"):
secs = float(parts[i + 1].strip("s,."))
global_rate_limit_expires = time.time() + secs
break
if p.lower() == "after":
secs = float(parts[i + 1].strip("s,."))
global_rate_limit_expires = time.time() + secs
break
else:
for p in parts:
try:
secs = float(p.strip("s,."))
if 0 < secs < 3600:
global_rate_limit_expires = time.time() + secs
break
except ValueError:
pass
continue
except Exception:
pass
# ---------------------------------------------------------------------------
# Shared modals
# ---------------------------------------------------------------------------
class ShuttlePane(Container):
"""Shuttle (migration) operations pane — clone, roles, emojis, metadata, messages, danger zone."""
class ProgressModal(ModalScreen[None]):
"""Modal to display progress for any long-running operation."""
def compose(self) -> ComposeResult:
with Vertical(id="progress_dialog"):
yield Label("Operation Status", id="progress_status")
yield ProgressBar(total=None, show_eta=False, id="progress_bar")
yield RichLog(id="progress_log", highlight=True, markup=True)
yield Button("Close", id="btn_close_progress", disabled=True)
def on_button_pressed(self, event: Button.Pressed):
if event.button.id == "btn_close_progress":
self.dismiss(None)
def write(self, message: str):
self.query_one("#progress_log", RichLog).write(message)
def set_status(self, status: str):
self.query_one("#progress_status", Label).update(status)
def set_progress(self, current: int, total: int):
bar = self.query_one("#progress_bar", ProgressBar)
bar.update(total=total, progress=current)
def allow_close(self):
btn = self.query_one("#btn_close_progress", Button)
btn.disabled = False
btn.variant = "success"
self.query_one("#progress_bar", ProgressBar).update(total=100, progress=100)
class ShuttleConfigModal(ModalScreen[dict]):
"""Modal for editing all shuttle-mode tokens & IDs."""
def __init__(self, config, target_platform: str):
super().__init__()
self.config = config
self.target_platform = target_platform
def compose(self) -> ComposeResult:
with Vertical(id="shuttle_config_dialog"):
yield Label("Shuttle Configuration", id="config_title")
yield Label("Discord Bot Token:")
yield Input(value=self.config.discord_bot_token or "", id="inp_d_token")
yield Label("Discord Server ID:")
yield Input(value=self.config.discord_server_id or "", id="inp_d_server")
plat_label = "Fluxer" if self.target_platform == "fluxer" else "Stoat"
yield Label(f"{plat_label} Bot Token:")
yield Input(value=self.config.target_bot_token or "", id="inp_t_token")
yield Label(f"{plat_label} Server/Community ID:")
yield Input(value=self.config.target_server_id or "", id="inp_t_server")
with Horizontal(id="config_buttons"):
yield Button("Save", variant="success", id="btn_save")
yield Button("Cancel", variant="primary", id="btn_cancel")
def on_button_pressed(self, event: Button.Pressed):
if event.button.id == "btn_save":
self.dismiss({
"d_token": self.query_one("#inp_d_token", Input).value,
"d_server": self.query_one("#inp_d_server", Input).value,
"t_token": self.query_one("#inp_t_token", Input).value,
"t_server": self.query_one("#inp_t_server", Input).value,
})
elif event.button.id == "btn_cancel":
self.dismiss(None)
class PlatformSelectModal(ModalScreen[str]):
"""Modal for selecting target platform (fluxer / stoat)."""
def compose(self) -> ComposeResult:
with Vertical(id="platform_select_dialog"):
yield Label("Select Target Platform", id="platform_title")
yield Button("Fluxer", variant="primary", id="btn_fluxer")
yield Button("Stoat", variant="warning", id="btn_stoat")
yield Rule()
yield Button("Cancel", id="btn_cancel_platform")
def on_button_pressed(self, event: Button.Pressed):
if event.button.id == "btn_fluxer":
self.dismiss("fluxer")
elif event.button.id == "btn_stoat":
self.dismiss("stoat")
elif event.button.id == "btn_cancel_platform":
self.dismiss(None)
class SubMenuModal(ModalScreen[str]):
"""A generic sub-menu modal that presents a list of labelled buttons."""
def __init__(self, title: str, options: list[tuple[str, str, str]]):
"""options: list of (button_id, label, variant)"""
super().__init__()
self._title = title
self._options = options
def compose(self) -> ComposeResult:
with Vertical(id="submenu_dialog"):
yield Label(self._title, id="submenu_title")
for btn_id, label, variant in self._options:
yield Button(label, id=btn_id, variant=variant)
yield Rule()
yield Button("Cancel", id="btn_cancel_sub")
def on_button_pressed(self, event: Button.Pressed):
if event.button.id == "btn_cancel_sub":
self.dismiss(None)
else:
self.dismiss(event.button.id)
class ConfirmModal(ModalScreen[bool]):
"""Simple Yes / No confirmation modal."""
def __init__(self, message: str, danger: bool = False):
super().__init__()
self._message = message
self._danger = danger
def compose(self) -> ComposeResult:
with Vertical(id="confirm_dialog"):
yield Label(self._message, id="confirm_msg")
with Horizontal(id="confirm_buttons"):
yield Button("Confirm", variant="error" if self._danger else "success", id="btn_yes")
yield Button("Cancel", variant="primary", id="btn_no")
def on_button_pressed(self, event: Button.Pressed):
self.dismiss(event.button.id == "btn_yes")
class ChannelPickerModal(ModalScreen[int]):
"""Modal listing Discord channels for single-channel selection."""
def __init__(self, channels: list, categories: dict, label: str = "Select Channel"):
super().__init__()
self._channels = channels
self._categories = categories
self._label = label
def compose(self) -> ComposeResult:
with Vertical(id="chanpick_dialog"):
yield Label(self._label, id="chanpick_title")
with VerticalScroll(id="chanpick_scroll"):
cat_grouped: dict[int | None, list] = {}
for c in self._channels:
cat_id = getattr(c, "category_id", None) if not isinstance(c, dict) else c.get("parent_id")
cat_grouped.setdefault(cat_id, []).append(c)
for cat_id in sorted(cat_grouped, key=lambda k: self._categories.get(k, "") if k else ""):
if cat_id is not None and cat_id in self._categories:
yield Label(f"[cyan]{self._categories[cat_id]}[/cyan]", classes="category_header")
for c in cat_grouped[cat_id]:
if isinstance(c, dict):
name = c.get("name", "Unnamed")
cid = c.get("id")
else:
name = c.name
cid = c.id
yield RadioButton(name, value=False, id=f"chpk_{cid}")
with Horizontal(id="chanpick_buttons"):
yield Button("Select", variant="success", id="btn_pick_ok")
yield Button("Cancel", id="btn_pick_cancel")
def on_button_pressed(self, event: Button.Pressed):
if event.button.id == "btn_pick_cancel":
self.dismiss(None)
elif event.button.id == "btn_pick_ok":
for rb in self.query(RadioButton):
if rb.value and rb.id and rb.id.startswith("chpk_"):
self.dismiss(int(rb.id.split("_", 1)[1]))
return
# Nothing selected
return
# ---------------------------------------------------------------------------
# ShuttleScreen — main screen for Shuttle Mode
# ---------------------------------------------------------------------------
class ShuttleScreen(Screen):
"""Native Textual screen for Shuttle (direct migration) mode."""
CSS = """
#shuttle_scroll {
align: center middle;
DEFAULT_CSS = """
ShuttlePane { height: auto; width: 100%; }
ShuttlePane #sp_info {
height: auto; border: tall cyan; padding: 1; margin-bottom: 1;
}
#shuttle_container {
width: 80%;
height: auto;
min-height: 25;
border: solid #4641D9;
padding: 1 2;
margin: 2 0;
}
#shuttle_title {
text-style: bold;
color: #4641D9;
margin-bottom: 1;
content-align: center middle;
width: 100%;
}
#shuttle_info {
height: auto;
margin-bottom: 2;
border: tall cyan;
padding: 1;
}
#shuttle_actions {
height: auto;
layout: vertical;
align: center top;
margin-top: 1;
}
#shuttle_actions Button {
width: 100%;
margin-bottom: 1;
}
/* Modals */
#shuttle_config_dialog, #platform_select_dialog, #submenu_dialog, #confirm_dialog {
width: 60%;
height: auto;
max-height: 80%;
border: thick $background 80%;
background: $surface;
padding: 1 2;
}
#progress_dialog {
width: 80%;
height: 80%;
border: thick $background 80%;
background: $surface;
padding: 1 2;
}
#chanpick_dialog {
width: 70%;
height: 75%;
border: thick $background 80%;
background: $surface;
padding: 1 2;
}
#chanpick_scroll {
height: 1fr;
border: solid $primary;
margin-bottom: 1;
padding: 0 1;
}
.category_header {
margin-top: 1;
background: $primary 10%;
text-style: bold;
padding-left: 1;
}
#config_title, #platform_title, #submenu_title, #confirm_msg, #chanpick_title {
text-style: bold; margin-bottom: 1;
}
#config_buttons, #confirm_buttons, #chanpick_buttons {
height: auto; margin-top: 1;
}
#config_buttons Button, #confirm_buttons Button, #chanpick_buttons Button {
width: 1fr; margin: 0 1;
}
#progress_status { text-style: bold; margin-bottom: 1; }
#progress_bar { margin-bottom: 1; }
#progress_log { height: 1fr; margin-bottom: 1; border: solid $primary; }
RadioButton:focus { background: transparent; border: none; }
RadioButton > .radio-button--label { padding: 0 1; }
RadioButton:focus > .radio-button--label { background: transparent; text-style: none; }
ShuttlePane #sp_actions { height: auto; }
ShuttlePane #sp_actions Button { width: 100%; margin-bottom: 1; }
"""
BINDINGS = [
("q", "app.exit", "Quit"),
("b", "go_back", "Back"),
]
def __init__(self, cfg_name: str, cfg_path: Path, *args, **kwargs):
super().__init__(*args, **kwargs)
self.cfg_name = cfg_name
self.config_path = cfg_path
self.config = load_config(self.config_path)
self.target_platform: str | None = None
self.config = load_config(cfg_path)
self.target_platform = self.config.target_platform or "fluxer"
self.engine: MigrationContext | None = None
self.validation_results: dict = {}
self.tokens_valid = False
self.permissions_complete = False
# Register rate-limit handler
rl = RateLimitHandler()
logging.getLogger("discord").addHandler(rl)
logging.getLogger("fluxer").addHandler(rl)
logging.getLogger("stoat").addHandler(rl)
def compose(self) -> ComposeResult:
with VerticalScroll():
with Vertical(id="sp_info"):
yield Label("Discord: [yellow]Loading...[/yellow]", id="sp_lbl_discord")
yield Label("Target: [yellow]Loading...[/yellow]", id="sp_lbl_target")
yield Label("Status: [yellow]Validating...[/yellow]", id="sp_lbl_status")
with Vertical(id="sp_actions"):
yield Button("Clone Server Template", id="sp_clone", disabled=True)
yield Button("Copy Roles & Permissions", id="sp_roles", disabled=True)
yield Button("Copy Emojis & Stickers", id="sp_emojis", disabled=True)
yield Button("Sync Server Profile", id="sp_metadata", disabled=True)
yield Button("Migrate Message History", id="sp_messages", disabled=True)
yield Rule()
yield Button("Danger Zone ⚠", id="sp_danger", variant="error", disabled=True)
# ── helpers ──────────────────────────────────────────────────────────
def on_mount(self) -> None:
self._rebuild_engine()
self.run_validate()
def reload_config(self) -> None:
self.config = load_config(self.config_path)
self.target_platform = self.config.target_platform or "fluxer"
self._rebuild_engine()
self.run_validate()
def _base_dir(self) -> str:
return f"Reaper-{self.cfg_name}"
@ -382,89 +129,51 @@ class ShuttleScreen(Screen):
def _rebuild_engine(self):
self.engine = MigrationContext(self.config, self.target_platform)
# ── labels ────────────────────────────────────────────────────────────
def _update_info_labels(self):
d_name = self.validation_results.get("discord_server_name")
d_disp = f"[green]\"{d_name}\"[/green]" if d_name else "[red]NOT SET UP[/red]"
self.query_one("#lbl_discord", Label).update(f"Discord: {d_disp}")
v = self.validation_results
if self.target_platform == "fluxer":
t_name = self.validation_results.get("target_community_name")
t_disp = f"[green]\"{t_name}\"[/green]" if t_name else "[red]NOT SET UP[/red]"
self.query_one("#lbl_target", Label).update(f"Fluxer: {t_disp}")
# Discord
d_name = v.get("discord_server_name")
d_bot = v.get("discord_bot_name")
if v.get("discord_timeout"):
d_disp = "[red]TIMEOUT[/red]"
elif d_name and v.get("discord_token") and v.get("discord_server"):
d_disp = f'[green]"{d_name}"[/green] Bot: [green]{d_bot}[/green]'
elif v.get("discord_token") is False:
d_disp = "[red]INVALID TOKEN[/red]"
else:
t_name = self.validation_results.get("target_community_name")
t_disp = f"[green]\"{t_name}\"[/green]" if t_name else "[red]NOT SET UP[/red]"
self.query_one("#lbl_target", Label).update(f"Stoat: {t_disp}")
d_disp = "[red]NOT SET UP[/red]"
self.query_one("#sp_lbl_discord", Label).update(f"Discord: {d_disp}")
# Target
plat = "Fluxer" if self.target_platform == "fluxer" else "Stoat"
t_name = v.get("target_community_name")
if v.get("target_timeout"):
t_disp = "[red]TIMEOUT[/red]"
elif t_name and v.get("target_token") and v.get("target_community"):
t_disp = f'[green]"{t_name}"[/green]'
elif v.get("target_token") is False:
t_disp = "[red]INVALID TOKEN[/red]"
else:
t_disp = "[red]NOT SET UP[/red]"
self.query_one("#sp_lbl_target", Label).update(f"{plat}: {t_disp}")
# Status
if not self.tokens_valid:
val = "[red][INVALID][/red]"
elif not self.permissions_complete:
val = "[yellow][PERMISSION MISSING][/yellow]"
val = "[yellow][MISSING PERMISSIONS][/yellow]"
else:
val = "[green][VALID][/green]"
self.query_one("#lbl_status", Label).update(f"Status: {val}")
self.query_one("#sp_lbl_status", Label).update(f"Status: {val}")
enabled = self.tokens_valid
for bid in ("#btn_clone", "#btn_roles", "#btn_emojis", "#btn_metadata", "#btn_messages", "#btn_danger"):
self.query_one(bid, Button).disabled = not enabled
# Buttons
for bid in ("#sp_clone", "#sp_roles", "#sp_emojis", "#sp_metadata", "#sp_messages", "#sp_danger"):
self.query_one(bid, Button).disabled = not self.tokens_valid
# ── compose ──────────────────────────────────────────────────────────
def compose(self) -> ComposeResult:
yield Header(show_clock=True)
with VerticalScroll(id="shuttle_scroll"):
with Container(id="shuttle_container"):
yield Label("Shuttle Mode", id="shuttle_title")
with Vertical(id="shuttle_info"):
yield Label("Discord: [yellow]Loading...[/yellow]", id="lbl_discord")
yield Label("Target: [yellow]Loading...[/yellow]", id="lbl_target")
yield Label("Status: [yellow]Validating...[/yellow]", id="lbl_status")
with Vertical(id="shuttle_actions"):
yield Button("Clone Server Template", id="btn_clone", disabled=True)
yield Button("Copy Roles & Permissions", id="btn_roles", disabled=True)
yield Button("Copy Emojis & Stickers", id="btn_emojis", disabled=True)
yield Button("Sync Server Profile", id="btn_metadata", disabled=True)
yield Button("Migrate Message History", id="btn_messages", disabled=True)
yield Rule()
yield Button("Configuration", id="btn_config")
yield Button("Danger Zone ⚠", id="btn_danger", variant="error", disabled=True)
yield Rule()
yield Button("Back", id="btn_back")
yield Footer()
# ── lifecycle ────────────────────────────────────────────────────────
def on_mount(self) -> None:
# Platform is already configured in config.yaml via ConfigScreen
self.target_platform = self.config.target_platform or "fluxer"
self._rebuild_engine()
self.run_validate()
def action_go_back(self):
self.app.pop_screen()
# ── button routing ───────────────────────────────────────────────────
def on_button_pressed(self, event: Button.Pressed):
bid = event.button.id
if bid == "btn_back":
self.app.pop_screen()
elif bid == "btn_config":
self._open_config()
elif bid == "btn_clone":
self.run_clone_template()
elif bid == "btn_roles":
self._open_roles_menu()
elif bid == "btn_emojis":
self._open_emoji_menu()
elif bid == "btn_metadata":
self._open_metadata_menu()
elif bid == "btn_messages":
self.run_migrate_messages()
elif bid == "btn_danger":
self._open_danger_menu()
# ── (0) validation ───────────────────────────────────────────────────
# ── validation ────────────────────────────────────────────────────────
@work(exclusive=True)
async def run_validate(self) -> None:
@ -501,7 +210,6 @@ class ShuttleScreen(Screen):
if all_tasks:
done, _ = await asyncio.wait(all_tasks, timeout=10.0)
# Discord
dt = tasks.get("discord")
if dt and dt in done:
res = dt.result()
@ -515,7 +223,6 @@ class ShuttleScreen(Screen):
self.validation_results["discord_timeout"] = True
dt.cancel()
# Target platform
tt = tasks.get("target")
if tt and tt in done:
res = tt.result()
@ -528,12 +235,10 @@ class ShuttleScreen(Screen):
self.validation_results["target_timeout"] = True
tt.cancel()
# Compute validity
discord_ok = self.validation_results.get("discord_token") and self.validation_results.get("discord_server")
target_ok = self.validation_results.get("target_token") and self.validation_results.get("target_community")
self.tokens_valid = bool(discord_ok and target_ok)
# Set state folder
if self.tokens_valid:
srv_id = self.config.target_server_id
srv_name = self.validation_results.get("target_community_name", "unknown")
@ -541,7 +246,6 @@ class ShuttleScreen(Screen):
safe = re.sub(r"[^a-zA-Z0-9_\-\.]", "_", srv_name)
self.engine.state.set_folder(str(srv_id), safe, base_dir=self._base_dir())
# Check permissions
self.permissions_complete = True
if self.tokens_valid:
di = self.validation_results.get("discord_intents", {})
@ -560,7 +264,26 @@ class ShuttleScreen(Screen):
self._update_info_labels()
# ── (1) clone server template ────────────────────────────────────────
# ── button routing ────────────────────────────────────────────────────
def on_button_pressed(self, event: Button.Pressed) -> None:
bid = event.button.id
if not bid or not bid.startswith("sp_"):
return
if bid == "sp_clone":
self.run_clone_template()
elif bid == "sp_roles":
self._open_roles_menu()
elif bid == "sp_emojis":
self._open_emoji_menu()
elif bid == "sp_metadata":
self._open_metadata_menu()
elif bid == "sp_messages":
self.run_migrate_messages()
elif bid == "sp_danger":
self._open_danger_menu()
# ── (1) clone server template ─────────────────────────────────────────
@work(exclusive=True)
async def run_clone_template(self) -> None:
@ -616,7 +339,7 @@ class ShuttleScreen(Screen):
modal.set_status("Finished.")
modal.allow_close()
# ── (2) roles & permissions ──────────────────────────────────────────
# ── (2) roles & permissions ───────────────────────────────────────────
def _open_roles_menu(self):
options = [
@ -707,7 +430,7 @@ class ShuttleScreen(Screen):
modal.set_status("Finished.")
modal.allow_close()
# ── (3) emojis & stickers ────────────────────────────────────────────
# ── (3) emojis & stickers ─────────────────────────────────────────────
def _open_emoji_menu(self):
options = [
@ -775,7 +498,7 @@ class ShuttleScreen(Screen):
modal.set_status("Finished.")
modal.allow_close()
# ── (4) server metadata sync ─────────────────────────────────────────
# ── (4) server metadata sync ──────────────────────────────────────────
def _open_metadata_menu(self):
options = [
@ -842,7 +565,7 @@ class ShuttleScreen(Screen):
modal.set_status("Finished.")
modal.allow_close()
# ── (5) message migration ────────────────────────────────────────────
# ── (5) message migration ─────────────────────────────────────────────
@work(exclusive=True)
async def run_migrate_messages(self) -> None:
@ -870,8 +593,8 @@ class ShuttleScreen(Screen):
modal.allow_close()
return
# Pick source channel via modal
self.app.pop_screen() # pop progress temporarily
# Pick source channel
self.app.pop_screen()
loop = asyncio.get_running_loop()
src_future = loop.create_future()
@ -889,7 +612,7 @@ class ShuttleScreen(Screen):
source_channel = next(c for c in d_channels if c.id == src_id)
# Pick target channel
# Fetch target channels
modal2_status = ProgressModal()
self.app.push_screen(modal2_status)
await asyncio.sleep(0.1)
@ -912,7 +635,7 @@ class ShuttleScreen(Screen):
if not recommended:
recommended = next((c for c in f_channels if c.get("name") == source_channel.name), None)
self.app.pop_screen() # pop status
self.app.pop_screen()
target_cat_names = {str(c.get("id")): c.get("name") for c in full_f if c.get("type") == 4}
@ -999,22 +722,7 @@ class ShuttleScreen(Screen):
modal.set_status("Finished.")
modal.allow_close()
# ── (6) configuration ────────────────────────────────────────────────
def _open_config(self):
def on_result(data: dict | None):
if data is None:
return
self.config.discord_bot_token = data["d_token"] or self.config.discord_bot_token
self.config.discord_server_id = data["d_server"] or self.config.discord_server_id
self.config.target_bot_token = data["t_token"] or self.config.target_bot_token
self.config.target_server_id = data["t_server"] or self.config.target_server_id
save_config(self.config, self.config_path)
self._rebuild_engine()
self.run_validate()
self.app.push_screen(ShuttleConfigModal(self.config, self.target_platform), on_result)
# ── (7) danger zone ──────────────────────────────────────────────────
# ── (6) danger zone ───────────────────────────────────────────────────
def _open_danger_menu(self):
options = [