I38/scripts/run_dialog.py

#!/usr/bin/env python3

# This file is part of I38.
#
# I38 is free software: you can redistribute it and/or modify it under the terms
# of the GNU General Public License as published by the Free Software
# Foundation, either version 3 of the License, or (at your option) any later
# version.
#
# I38 is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE. See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along with
# I38. If not, see <https://www.gnu.org/licenses/>.

import argparse
import os
import shutil
import subprocess
from dataclasses import dataclass
from pathlib import Path
from urllib.parse import urlparse

import gi

gi.require_version("Atk", "1.0")
gi.require_version("Gdk", "3.0")
gi.require_version("Gio", "2.0")
gi.require_version("GLib", "2.0")
gi.require_version("Gtk", "3.0")
from gi.repository import Atk, Gdk, Gio, GLib, Gtk


APP_NAME = "I38 Run Dialog"
MAX_HISTORY_ENTRIES = 50
MAX_RESULTS = 100
PATH_PREFIXES = ("/", "./", "../", "~/")
RESULT_FEEDBACK_DELAY_MS = 150
RESULT_SOUND = ["play", "-nqV0", "synth", "0.05", "saw", "300:1200", "vol", "0.20"]


@dataclass(frozen=True)
class Result:
	kind: str
	label: str
	detail: str
	value: str
	app_id: str = ""


def state_dir():
	value = os.environ.get("XDG_STATE_HOME")
	if value:
		return Path(value) / "I38"
	return Path.home() / ".local" / "state" / "I38"


def history_path():
	return state_dir() / "run-dialog-history"


def legacy_history_path():
	return Path(__file__).resolve().parent / ".history"


def migrate_history():
	path = history_path()
	if path.exists() or not legacy_history_path().is_file():
		return

	path.parent.mkdir(parents=True, exist_ok=True)
	shutil.copyfile(legacy_history_path(), path)


def read_history():
	try:
		return [
			line.strip()
			for line in history_path().read_text(encoding="utf-8").splitlines()
			if line.strip()
		][:MAX_HISTORY_ENTRIES]
	except OSError:
		return []


def write_history(entry, items):
	entry = entry.strip()
	if not entry:
		return items[:MAX_HISTORY_ENTRIES]

	updated_items = [entry]
	updated_items.extend(item for item in items if item != entry)
	return updated_items[:MAX_HISTORY_ENTRIES]


def save_history(items):
	path = history_path()
	path.parent.mkdir(parents=True, exist_ok=True)
	path.write_text("\n".join(items[:MAX_HISTORY_ENTRIES]) + "\n", encoding="utf-8")


def path_commands():
	commands = set()
	for directory in os.environ.get("PATH", "").split(os.pathsep):
		if not directory:
			continue
		try:
			for item in os.scandir(directory):
				if item.is_file() and os.access(item.path, os.X_OK):
					commands.add(item.name)
		except OSError:
			continue
	return sorted(commands, key=str.casefold)


def is_path_query(text):
	return text.startswith(PATH_PREFIXES)


def path_results(text):
	if not is_path_query(text):
		return []

	expanded_text = os.path.expanduser(text)
	if expanded_text.endswith(os.sep):
		directory = Path(expanded_text)
		name_prefix = ""
		display_parent = text
	else:
		directory = Path(expanded_text).parent
		name_prefix = Path(expanded_text).name
		display_parent = text[: len(text) - len(name_prefix)]

	try:
		items = list(os.scandir(directory))
	except OSError:
		return []

	results = []
	for item in items:
		if not item.name.casefold().startswith(name_prefix.casefold()):
			continue

		value = f"{display_parent}{item.name}"
		try:
			if item.is_dir():
				results.append(Result("directory", f"{item.name}/", "Directory", f"{value}/"))
			elif item.is_file() and os.access(item.path, os.X_OK):
				results.append(Result("file", item.name, "Executable file", value))
			else:
				results.append(Result("file", item.name, "File", value))
		except OSError:
			continue

	return sorted(
		results,
		key=lambda item: (item.kind != "directory", item.label.casefold()),
	)[:MAX_RESULTS]


def matches(query, value):
	return query.casefold() in value.casefold()


def application_results(query, applications):
	results = []
	for app in applications:
		name = app.get_display_name() or app.get_name()
		executable = app.get_executable() or ""
		if not executable:
			continue
		if not matches(query, name) and not matches(query, executable):
			continue
		results.append(Result("application", name, f"Application: {executable}", executable, app.get_id() or ""))

	return sorted(results, key=lambda item: item.label.casefold())


def command_results(query, commands):
	return [
		Result("command", command, "Command", command)
		for command in commands
		if matches(query, command)
	]


def history_results(query, items):
	return [
		Result("history", item, "History", item)
		for item in items
		if matches(query, item)
	]


def merge_results(*groups):
	results = []
	seen = set()
	for group in groups:
		for result in group:
			key = (result.kind, result.value, result.app_id)
			if key in seen:
				continue
			seen.add(key)
			results.append(result)
			if len(results) >= MAX_RESULTS:
				return results
	return results


def normalized_path(text):
	text = text.strip()
	if len(text) >= 2 and text[0] == text[-1] and text[0] in ("'", '"'):
		text = text[1:-1]
	return Path(os.path.expanduser(text))


def raw_target_kind(text):
	text = text.strip()
	parsed = urlparse(text)
	if parsed.scheme in ("ftp", "http", "https", "file", "mailto"):
		return "uri"
	if text.startswith("www."):
		return "uri"
	if text.startswith("man://"):
		return "manual"
	if normalized_path(text).exists():
		return "path"
	return "command"


def result_count_message(count):
	return f"{count} suggestion" if count == 1 else f"{count} suggestions"


def play_result_sound():
	try:
		subprocess.Popen(
			RESULT_SOUND,
			stdout=subprocess.DEVNULL,
			stderr=subprocess.DEVNULL,
			start_new_session=True,
		)
	except OSError:
		pass


class RunDialog(Gtk.Window):
	def __init__(self):
		super().__init__(title=APP_NAME)
		self.set_default_size(720, 420)
		self.set_border_width(12)
		self.set_position(Gtk.WindowPosition.CENTER)
		self.connect("destroy", Gtk.main_quit)
		self.connect("key-press-event", self.on_window_key_press)

		migrate_history()
		self.history = read_history()
		self.commands = path_commands()
		self.applications = {
			app.get_id(): app
			for app in Gio.AppInfo.get_all()
			if app.should_show() and app.get_id()
		}
		self.results = []
		self.last_feedback_count = None
		self.pending_feedback_count = None
		self.feedback_timeout_id = None

		main_box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=8)
		self.add(main_box)

		entry_label = Gtk.Label(label="_Command, application, or file:")
		entry_label.set_xalign(0)
		main_box.pack_start(entry_label, False, False, 0)

		self.entry = Gtk.Entry()
		self.entry.set_placeholder_text("Type a command, application name, or path")
		self.entry.connect("changed", self.on_entry_changed)
		self.entry.connect("activate", self.on_entry_activate)
		self.entry.connect("key-press-event", self.on_entry_key_press)
		entry_label.set_mnemonic_widget(self.entry)
		main_box.pack_start(self.entry, False, False, 0)

		self.store = Gtk.ListStore(str, str)
		self.view = Gtk.TreeView(model=self.store)
		self.view.set_headers_visible(True)
		self.view.connect("row-activated", self.on_row_activated)
		self.view.connect("key-press-event", self.on_view_key_press)
		self.view.get_selection().set_mode(Gtk.SelectionMode.SINGLE)

		result_column = Gtk.TreeViewColumn("Result", Gtk.CellRendererText(), text=0)
		detail_column = Gtk.TreeViewColumn("Type", Gtk.CellRendererText(), text=1)
		self.view.append_column(result_column)
		self.view.append_column(detail_column)

		scrolled = Gtk.ScrolledWindow()
		scrolled.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC)
		scrolled.add(self.view)
		main_box.pack_start(scrolled, True, True, 0)

		self.status = Gtk.Statusbar()
		self.status_context_id = self.status.get_context_id("run-dialog-results")
		main_box.pack_start(self.status, False, False, 0)

		button_box = Gtk.ButtonBox(orientation=Gtk.Orientation.HORIZONTAL)
		button_box.set_layout(Gtk.ButtonBoxStyle.END)
		open_button = Gtk.Button.new_with_mnemonic("_Open")
		open_button.connect("clicked", self.on_open_clicked)
		cancel_button = Gtk.Button.new_with_mnemonic("_Cancel")
		cancel_button.connect("clicked", lambda _button: self.close())
		button_box.add(cancel_button)
		button_box.add(open_button)
		main_box.pack_start(button_box, False, False, 0)

		window_accessible = self.get_accessible()
		window_accessible.set_name(APP_NAME)
		window_accessible.set_description(
			"Type to find commands and applications. Type a path to browse files. "
			"Use up and down arrows to select a result, Tab to insert the selected result, "
			"and Enter to open it."
		)
		self.entry.get_accessible().set_name("Command, application, or file")
		self.view.get_accessible().set_name("Run dialog suggestions")
		self.view.get_accessible().set_role(Atk.Role.LIST)
		self.status.get_accessible().set_name("Run dialog status")

		self.refresh_results()
		self.set_status("Type to search. Enter a path such as ~/doomtmp/ to browse files.")
		GLib.idle_add(self.entry.grab_focus)

	def refresh_results(self):
		query = self.entry.get_text().strip()
		if is_path_query(query):
			self.results = path_results(query)
		else:
			self.results = merge_results(
				history_results(query, self.history),
				application_results(query, self.applications.values()),
				command_results(query, self.commands),
			)

		self.store.clear()
		for result in self.results:
			self.store.append((result.label, result.detail))

		self.view.get_selection().unselect_all()
		self.schedule_result_feedback(len(self.results))

	def set_status(self, message):
		self.status.remove_all(self.status_context_id)
		self.status.push(self.status_context_id, message)

	def schedule_result_feedback(self, count):
		if self.last_feedback_count is None:
			self.last_feedback_count = count
			return
		if count == self.pending_feedback_count:
			return

		self.pending_feedback_count = count
		if self.feedback_timeout_id is not None:
			GLib.source_remove(self.feedback_timeout_id)
		self.feedback_timeout_id = GLib.timeout_add(
			RESULT_FEEDBACK_DELAY_MS,
			self.announce_result_count,
		)

	def announce_result_count(self):
		self.feedback_timeout_id = None
		count = self.pending_feedback_count
		if count == self.last_feedback_count:
			return False

		self.last_feedback_count = count
		play_result_sound()
		message = result_count_message(count)
		self.set_status(message)
		status_accessible = self.status.get_accessible()
		status_accessible.emit("notification", message, Atk.Live.POLITE)
		return False

	def selected_result(self):
		model, tree_iter = self.view.get_selection().get_selected()
		if tree_iter is None:
			return None
		return self.results[model.get_path(tree_iter).get_indices()[0]]

	def insert_selected_result(self):
		result = self.selected_result()
		if result is None:
			return
		self.entry.set_text(result.value)
		self.entry.set_position(-1)

	def remember_history(self, entry):
		entry = entry.strip()
		if not entry:
			return
		self.history = write_history(entry, self.history)
		save_history(self.history)

	def forget_history(self, entry):
		entry = entry.strip()
		if entry not in self.history:
			return
		self.history.remove(entry)
		save_history(self.history)

	def activate(self, result=None):
		if result is not None and result.kind == "directory":
			self.remember_history(result.value)
			self.entry.set_text(result.value)
			self.entry.set_position(-1)
			return

		text = result.value if result is not None else self.entry.get_text().strip()
		if not text:
			return

		if result is not None and result.kind == "application":
			self.launch_application(result)
			return
		self.launch_raw(text)

	def launch_application(self, result):
		app = self.applications.get(result.app_id)
		if app is None:
			self.show_error(f"Application is no longer available: {result.label}")
			return
		try:
			launched = app.launch([], None)
		except GLib.Error as error:
			self.forget_history(result.value)
			self.show_error(f"Could not open {result.label}.\n\n{error.message}")
			return
		if not launched:
			self.forget_history(result.value)
			self.show_error(f"Could not open {result.label}.")
			return
		self.remember_history(result.value)
		self.close()

	def launch_raw(self, text):
		kind = raw_target_kind(text)
		if kind == "uri":
			target = f"https://{text}" if text.startswith("www.") else text
			command = ["xdg-open", target]
		elif kind == "manual":
			page = text.removeprefix("man://")
			if not page:
				self.show_error("Enter a manual page after man://")
				return
			command = [
				"bash",
				"-o",
				"pipefail",
				"-c",
				'man -- "$1" | yad --text-info --show-cursor --button "Close:0" --title "I38 manual page" -',
				"_",
				page,
			]
		elif kind == "path":
			path = normalized_path(text)
			if path.is_file() and os.access(path, os.X_OK):
				command = [str(path)]
			else:
				command = ["xdg-open", str(path)]
		else:
			command = ["bash", "-c", text]

		self.start_command(command, text)

	def start_command(self, command, description):
		try:
			process = subprocess.Popen(
				command,
				stdout=subprocess.DEVNULL,
				stderr=subprocess.DEVNULL,
				text=True,
				start_new_session=True,
			)
		except OSError as error:
			self.show_error(f"Could not open {description}.\n\n{error}")
			return

		self.hide()
		GLib.timeout_add(250, self.check_process, process, description)

	def check_process(self, process, description):
		status = process.poll()
		if status is None:
			self.remember_history(description)
			Gtk.main_quit()
			return False
		if status == 0:
			self.remember_history(description)
			Gtk.main_quit()
			return False

		self.forget_history(description)
		self.show_all()
		self.entry.grab_focus()
		self.show_error(
			f"Could not open {description}.\n\n"
			"No additional error details were provided."
		)
		return False

	def show_error(self, message):
		dialog = Gtk.MessageDialog(
			transient_for=self,
			flags=Gtk.DialogFlags.MODAL,
			message_type=Gtk.MessageType.ERROR,
			buttons=Gtk.ButtonsType.CLOSE,
			text=message,
		)
		dialog.set_title(APP_NAME)
		dialog.run()
		dialog.destroy()

	def on_entry_changed(self, _entry):
		self.refresh_results()

	def on_entry_activate(self, _entry):
		self.activate(self.selected_result())

	def on_open_clicked(self, _button):
		self.activate(self.selected_result())

	def on_row_activated(self, _view, path, _column):
		self.activate(self.results[path.get_indices()[0]])

	def on_entry_key_press(self, _entry, event):
		if event.keyval in (Gdk.KEY_Down, Gdk.KEY_Up):
			self.view.grab_focus()
			if self.results:
				if event.keyval == Gdk.KEY_Up:
					self.view.get_selection().select_path(len(self.results) - 1)
				else:
					self.view.get_selection().select_path(0)
			return True
		if event.keyval == Gdk.KEY_Tab:
			self.insert_selected_result()
			return True
		return False

	def on_view_key_press(self, _view, event):
		if event.keyval == Gdk.KEY_BackSpace:
			self.entry.grab_focus()
			self.entry.set_position(-1)
			self.entry.backspace()
			return True

		if event.keyval == Gdk.KEY_Tab:
			self.insert_selected_result()
			self.entry.grab_focus()
			return True

		if event.string and event.string.isprintable():
			self.entry.grab_focus()
			self.entry.set_position(-1)
			self.entry.insert_text(event.string, -1)
			return True

		return False

	def on_window_key_press(self, _window, event):
		if event.keyval == Gdk.KEY_Escape:
			self.close()
			return True
		return False


def run_self_test():
	class FakeApplication:
		def get_display_name(self):
			return "Mozilla Firefox"

		def get_name(self):
			return "Mozilla Firefox"

		def get_executable(self):
			return "/usr/lib/firefox/firefox"

		def get_id(self):
			return "firefox.desktop"

	test_dir = Path(os.environ["I38_RUN_DIALOG_TEST_DIR"])
	(test_dir / "folder").mkdir(parents=True)
	(test_dir / "tool").write_text("#!/bin/sh\n", encoding="utf-8")
	(test_dir / "note.txt").write_text("note\n", encoding="utf-8")
	(test_dir / "tool").chmod(0o755)

	results = path_results(f"{test_dir}/")
	assert [result.label for result in results] == ["folder/", "note.txt", "tool"]
	assert results[2].detail == "Executable file"
	assert raw_target_kind("mailto:test@example.com") == "uri"
	assert raw_target_kind("man://bash") == "manual"
	assert raw_target_kind(str(test_dir / "note.txt")) == "path"
	assert raw_target_kind('python3 "/tmp/file with spaces.py"') == "command"
	assert application_results("firef", [FakeApplication()])[0].value == "/usr/lib/firefox/firefox"
	assert merge_results(
		[Result("history", "firefox", "History", "firefox")],
		[Result("command", "firefox", "Command", "firefox")],
	)[0].detail == "History"
	assert result_count_message(0) == "0 suggestions"
	assert result_count_message(1) == "1 suggestion"
	assert result_count_message(12) == "12 suggestions"
	print("run dialog self-test passed")


def main():
	parser = argparse.ArgumentParser(description=APP_NAME)
	parser.add_argument("--self-test", action="store_true")
	args = parser.parse_args()

	if args.self_test:
		run_self_test()
		return

	window = RunDialog()
	window.show_all()
	Gtk.main()


if __name__ == "__main__":
	main()