content-media-image-picker/random_file_picker/tui.py

"""
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"📄 {file_count_str} {self.current_file}")
        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()