446 lines
17 KiB
Python
446 lines
17 KiB
Python
"""
|
|
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 collections.abc import Callable
|
|
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()
|
|
|
|
# Map valid sort keys to the function that extracts the comparison value
|
|
SORT_KEYS: dict[str, Callable[[pathlib.Path], float | int | str]] = {
|
|
"name": lambda p: p.name.lower(),
|
|
"date": lambda p: p.stat().st_mtime,
|
|
"size": lambda p: p.stat().st_size,
|
|
}
|
|
|
|
# Map valid order strings to their `reverse` boolean flag (True = descending)
|
|
SORT_ORDERS: dict[str, bool] = {
|
|
"a": False, "asc": False, "ascending": False,
|
|
"d": True, "desc": True, "descending": True,
|
|
}
|
|
|
|
def parse_and_sort_files(files: list[pathlib.Path], sort_args: list[str]) -> None:
|
|
"""Sort files in place and raise ValueError on invalid rules."""
|
|
rules: list[str] = []
|
|
for arg in sort_args:
|
|
rules.extend(r.strip() for r in arg.split(",") if r.strip())
|
|
|
|
for rule in reversed(rules):
|
|
parts = rule.split(":")
|
|
key_str = parts[0].strip().lower()
|
|
order_str = parts[1].strip().lower() if len(parts) > 1 else "a"
|
|
|
|
if key_str not in SORT_KEYS:
|
|
raise ValueError(f"Invalid sort key '{key_str}'. Valid options: {', '.join(SORT_KEYS.keys())}")
|
|
if order_str not in SORT_ORDERS:
|
|
raise ValueError(f"Invalid sort order '{order_str}'. Valid options: {', '.join(SORT_ORDERS.keys())}")
|
|
files.sort(key=SORT_KEYS[key_str], reverse=SORT_ORDERS[order_str])
|
|
|
|
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")
|
|
sort: list[str] | None = tap.arg("--sort", nargs="*", help="Sort rules (e.g. 'date:d, name:a'). Keys: name, date, size. Orders: a, d.")
|
|
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,
|
|
sort: list[str] | None = None,
|
|
):
|
|
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.sort_rules: list[str] = sort or []
|
|
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 self.sort_rules:
|
|
try:
|
|
parse_and_sort_files(self.usable_files, self.sort_rules)
|
|
except ValueError as e:
|
|
status.update(f"❌ Sorting error: {e}")
|
|
return
|
|
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
|
|
|
|
# Validate sort arguments before launching the TUI
|
|
if args.sort:
|
|
try:
|
|
parse_and_sort_files([], args.sort)
|
|
except ValueError as e:
|
|
print(f"❌ Sorting error: {e}")
|
|
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,
|
|
sort=args.sort or [], # default filesystem order if no sort rules are provided
|
|
)
|
|
app.run()
|
|
|
|
def main() -> None:
|
|
tap.Parser(Args).bind(main_with_tap).run()
|
|
|
|
if __name__ == "__main__":
|
|
main()
|