""" A script to randomly select files from a directory, open them, and allow the user to upload them to a Discord webhook, delete them, or skip to the next file. You can specify a file size limit, whether to descend into subdirectories, and provide blacklist/whitelist patterns for file selection. Here's an example usage: ./rand.py -s 5 -d /path/to/directory --descend --blacklist "*.png" "*.mp4" """ import pathlib import random from typing import ClassVar, cast, final, override import typed_argparse as tap from textual import work from textual.app import App, ComposeResult from textual.binding import BindingType from textual.containers import Grid from textual.screen import ModalScreen from textual.widgets import Footer, Header, Input, Label from textual.worker import Worker, WorkerState from random_file_picker.core import IMG_EXTENSIONS, collect_files, move_file_to_move_folder, trash_file, upload_file from random_file_picker.openers import ( is_geeqie_available, open_file_in_dolphin, open_file_with_default_app, open_file_with_geeqie, ) GEEQIE_AVAILABLE = is_geeqie_available() class Args(tap.TypedArgs): file_size: int = tap.arg("-s", default=10, help="Maximum file size in MB") directory: str = tap.arg("-d", default=".", help="Directory to search for files") recursive: bool = tap.arg("-r", "--recursive", "--descend", default=False, help="Whether to recurse into subdirectories") use_geeqie: bool = tap.arg("--use-geeqie", default=False, help="Whether to use Geeqie as the image viewer") no_trash: bool = tap.arg("--no-trash", default=False, help="Delete files immediately without moving to trash") no_random: bool = tap.arg("--no-random", default=False, help="Do not randomize the order of files") blacklist: list[str] | None = tap.arg("--blacklist", nargs="*", help="List of glob patterns to blacklist files") whitelist: list[str] | None = tap.arg("--whitelist", nargs="*", help="List of glob patterns to whitelist files") @final class JumpModal(ModalScreen[int | None]): """Modal screen to prompt the user for an item index to skip to.""" # css in the terminal before gta6 DEFAULT_CSS = """ JumpModal { align: center middle; } #dialog { grid-size: 1; padding: 1 2; width: 40; height: 11; border: thick $background 80%; background: $surface; } """ @override def compose(self) -> ComposeResult: yield Grid( Label("Enter item index to jump to:"), Input(placeholder="e.g. 5", id="index-input", type="integer"), id="dialog", ) def on_input_submitted(self, event: Input.Submitted) -> None: """Handle the submission of the input field, dismissing the modal with the entered index.""" val = event.value.strip() if val.isdigit(): _ = self.dismiss(int(val)) return _ = self.dismiss(None) @final class MenuApp(App[None]): TITLE = "File Picker" # This defines the keys shown in the footer BINDINGS: ClassVar[list[BindingType]] = [ ("left", "prev_file", "Previous File"), ("right", "next_file", "Next File"), ("r", "reopen_file", "Reopen"), ("colon", "jump_to_item", "Jump to Item"), ("g", "jump_to_item", "Jump to Item"), ("u", "upload", "Upload"), ("d", "delete", "Delete"), ("m", "move_to_folder", "Move File"), ("f", "open_folder", "Open Folder"), ("q", "quit", "Quit"), ] def __init__( self, *, directory: str = ".", file_limit: int = 10, recurse: bool = False, blacklist: list[str] | None = None, whitelist: list[str] | None = None, use_geeqie: bool = False, use_trash: bool = True, no_random: bool = False, ): super().__init__() self.directory: pathlib.Path = pathlib.Path(directory) self.file_limit: int = file_limit self.recurse: bool = recurse self.blacklist_patterns: list[str] = blacklist or [] self.whitelist_patterns: list[str] = whitelist or [] self.use_geeqie: bool = use_geeqie self.use_trash: bool = use_trash self.no_random: bool = no_random self.usable_files: list[pathlib.Path] = [] self.current_index: int = 0 self.current_file: pathlib.Path | None = None self.is_loading: bool = False @override def compose(self) -> ComposeResult: yield Header() yield Label("Loading files...", id="status") yield Label("", id="current-file") yield Footer() def on_mount(self) -> None: """On application mount, start gathering files in the background.""" self.start_gathering_files() @override def check_action(self, action: str, parameters: tuple[object, ...]) -> bool | None: """Disable every action except quit while files are being gathered.""" if self.is_loading and action != "quit": return None return True def start_gathering_files(self) -> None: """Start file collection in a background worker so the UI stays responsive.""" if not self.directory.exists() or not self.directory.is_dir(): self.query_one("#status", Label).update(f"❌ Directory '{self.directory}' not found") return self.is_loading = True self.refresh_bindings() self.query_one("#current-file", Label).update(f"🔍 Scanning '{self.directory}' for files, please wait...") status = self.query_one("#status", Label) status.update("") status.loading = True _ = self.gather_files_worker( self.directory, self.file_limit, self.blacklist_patterns, self.whitelist_patterns, recurse=self.recurse, ) @work(thread=True, exclusive=True) def gather_files_worker( self, directory: pathlib.Path, file_limit: int, blacklist_patterns: list[str], whitelist_patterns: list[str], *, recurse: bool, ) -> list[pathlib.Path]: """Run file collection in a worker thread so a large/recursive scan doesn't block the UI.""" return collect_files( search_path=directory, search_subdirectories=recurse, file_size_limit=file_limit, blacklist_patterns=blacklist_patterns, whitelist_patterns=whitelist_patterns, ) def on_worker_state_changed(self, event: Worker.StateChanged) -> None: """Handle the completion of the background file scan, updating the UI with results or errors.""" worker = cast("Worker[list[pathlib.Path]]", event.worker) if worker.name != "gather_files_worker": return status = self.query_one("#status", Label) self.is_loading = False status.loading = False self.refresh_bindings() if event.state == WorkerState.SUCCESS: files = worker.result or [] self.usable_files = list(files) if not self.no_random: random.shuffle(self.usable_files) if not self.usable_files: status.update("❌ No usable files found") else: status.update(f"✅ Found {len(self.usable_files)} files") self.current_index = 0 self.display_current_file() elif event.state == WorkerState.ERROR: status.update(f"❌ Failed to scan directory: {worker.error}") def assert_current_file_exists(self) -> bool: """Check if the current file exists and update the status label accordingly.""" if not self.current_file or not self.current_file.exists(): self.query_one("#status", Label).update(f"❌ File '{self.current_file}' not found") return False return True def display_current_file(self) -> None: """Display the current file based on current_index.""" if not self.usable_files or self.current_index >= len(self.usable_files): self.query_one("#status", Label).update("❌ No files available") self.current_file = None return self.current_file = self.usable_files[self.current_index] file_count_str = f"({self.current_index + 1}/{len(self.usable_files)})" self.query_one("#current-file", Label).update(f"📄 {self.current_file} {file_count_str}") self.open_file() def open_file(self) -> None: """Open the current file - display images in terminal, play others externally.""" if not self.current_file: return if not self.assert_current_file_exists(): return self.query_one("#status", Label).update(f"📝 Open {self.current_file}") try: if self.use_geeqie and str(self.current_file).lower().endswith(tuple(IMG_EXTENSIONS)): open_file_with_geeqie(self.current_file) else: ok = open_file_with_default_app(self.current_file) if not ok: self.query_one("#status", Label).update("❌ Failed to open file with default app") except Exception as e: self.query_one("#status", Label).update(f"❌ Failed to open file: {str(e)}") def send_file(self) -> None: """Upload the current file to Discord webhook.""" if not self.current_file: return if not self.assert_current_file_exists(): return try: self.query_one("#status", Label).update(f"⏳ Uploading {self.current_file.name} to webhook...") upload_file(self.current_file) self.query_one("#status", Label).update(f"✅ Uploaded {self.current_file.name} to webhook") except Exception as e: self.query_one("#status", Label).update(f"❌ Upload failed: {str(e)}") def delete_file(self) -> None: """Delete the current file (move to trash if enabled).""" if not self.current_file or self.current_index >= len(self.usable_files): return if not self.assert_current_file_exists(): return try: if self.use_trash: trash_file(self.current_file) else: self.current_file.unlink() _ = self.usable_files.pop(self.current_index) self.query_one("#status", Label).update("✅ Deleted file") # Adjust index if we deleted the last file if self.current_index >= len(self.usable_files) and self.usable_files: self.current_index = len(self.usable_files) - 1 if self.usable_files: self.display_current_file() else: self.query_one("#status", Label).update("❌ No more files") except Exception as e: self.query_one("#status", Label).update(f"❌ Delete failed: {str(e)}") def move_to_move_folder(self) -> None: """Move the current file to the move folder.""" if not self.current_file or self.current_index >= len(self.usable_files): return if not self.assert_current_file_exists(): return try: move_file_to_move_folder(self.current_file) _ = self.usable_files.pop(self.current_index) self.query_one("#status", Label).update("✅ Moved to move folder") # Adjust index if we removed the last file if self.current_index >= len(self.usable_files) and self.usable_files: self.current_index = len(self.usable_files) - 1 if self.usable_files: self.display_current_file() else: self.query_one("#status", Label).update("❌ No more files") except Exception as e: self.query_one("#status", Label).update(f"❌ Move failed: {str(e)}") def open_in_folder(self) -> None: """Open the current file's folder in the file manager.""" if not self.current_file: return if not self.assert_current_file_exists(): return open_file_in_dolphin(self.current_file) self.query_one("#status", Label).update("📂 Opened file in folder") def action_upload(self) -> None: """Upload the current file to Discord webhook.""" self.send_file() def action_delete(self) -> None: """Delete the current file (move to trash if enabled).""" self.delete_file() def action_move_to_folder(self) -> None: """Move the current file to the move folder.""" self.move_to_move_folder() def action_open_folder(self) -> None: """Open the current file's folder in the file manager.""" self.open_in_folder() def action_next_file(self) -> None: """Display the next file.""" if not self.usable_files: return self.current_index = (self.current_index + 1) % len(self.usable_files) self.display_current_file() def action_prev_file(self) -> None: """Display the previous file.""" if not self.usable_files: return self.current_index = (self.current_index - 1) % len(self.usable_files) self.display_current_file() def action_reopen_file(self) -> None: """Reopen the current file.""" if not self.current_file: return self.open_file() self.query_one("#status", Label).update("🔄 Reopened file") def action_jump_to_item(self) -> None: """Prompt for item number and jump to it.""" if not self.usable_files: return def check_jump(target_1based: int | None) -> None: if target_1based is None: return if 1 <= target_1based <= len(self.usable_files): self.current_index = target_1based - 1 self.display_current_file() else: self.query_one("#status", Label).update(f"❌ Index out of range (1 - {len(self.usable_files)})") _ = self.push_screen(JumpModal(), check_jump) def main_with_tap(args: Args) -> None: if args.use_geeqie and not GEEQIE_AVAILABLE: print("❌ Geeqie is not available on this system. Please install it or run without --use-geeqie.") return app = MenuApp( directory=args.directory, file_limit=args.file_size, recurse=args.recursive, blacklist=args.blacklist or [], whitelist=args.whitelist or [], use_geeqie=args.use_geeqie, use_trash=not args.no_trash, no_random=args.no_random, ) app.run() def main() -> None: tap.Parser(Args).bind(main_with_tap).run() if __name__ == "__main__": main()