I38/scripts/i38-clipboard.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# 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 socket
import sys
from pathlib import Path

import gi

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


APP_ID = "i38-clipboard"
APP_NAME = "I38 Clipboard History"
MAX_ELEMENTS = 50
MAX_ITEM_LENGTH = 80
SOCKET_MESSAGE_SHOW = b"show\n"
SOCKET_MESSAGE_PING = b"ping\n"


def data_dir():
	value = os.environ.get("XDG_DATA_HOME")
	if value:
		return Path(value) / APP_ID
	return Path.home() / ".local" / "share" / APP_ID


def runtime_dir():
	value = os.environ.get("XDG_RUNTIME_DIR")
	if value:
		return Path(value)
	return Path("/tmp")


def socket_path():
	return runtime_dir() / f"{APP_ID}-{os.getuid()}.sock"


def history_path():
	return data_dir() / "history"


def format_item(item):
	text = item.replace("\n", " ").replace("\t", " ")
	if len(text) <= MAX_ITEM_LENGTH:
		return text

	front_length = MAX_ITEM_LENGTH // 2
	back_length = max(1, MAX_ITEM_LENGTH - front_length - 3)
	return text[:front_length] + "..." + text[-back_length:]


class History:
	def __init__(self):
		self.items = []
		self.changed_callbacks = []

	def connect_changed(self, callback):
		self.changed_callbacks.append(callback)

	def notify_changed(self):
		for callback in self.changed_callbacks:
			callback(self.items)

	def add(self, item, is_from_selection=False):
		if item is None:
			return

		item = str(item)
		if item == "":
			return

		if item in self.items:
			self.items.remove(item)

		last_item = self.items[0] if self.items else None
		if is_from_selection and last_item is not None and (
			item.startswith(last_item) or item.endswith(last_item)
		):
			self.items[0] = item
		else:
			self.items.insert(0, item)

		if len(self.items) > MAX_ELEMENTS:
			self.items = self.items[:MAX_ELEMENTS]

		self.save()
		self.notify_changed()

	def clear(self):
		self.items = []
		self.save()
		self.notify_changed()

	def load(self):
		path = history_path()
		try:
			with path.open("rb") as file:
				length = file.readline()
				while length:
					try:
						bytes_to_read = int(length)
					except ValueError:
						break
					self.items.append(file.read(bytes_to_read).decode("UTF-8"))
					file.read(1)
					length = file.readline()
		except OSError:
			pass

	def save(self):
		path = history_path()
		try:
			path.parent.mkdir(parents=True, exist_ok=True)
			with path.open("wb") as file:
				for item in self.items:
					encoded = item.encode("UTF-8")
					file.write(str(len(encoded)).encode("UTF-8") + b"\n")
					file.write(encoded + b"\n")
		except OSError:
			pass


class ClipboardWatcher:
	def __init__(self, history):
		self.history = history
		self.default_clipboard = Gtk.Clipboard.get(Gdk.SELECTION_CLIPBOARD)
		self.primary_clipboard = Gtk.Clipboard.get(Gdk.SELECTION_PRIMARY)
		self.default_text = self.default_clipboard.wait_for_text()
		self.primary_text = self.primary_clipboard.wait_for_text()
		self.default_clipboard.connect("owner-change", self.on_default_owner_change)
		self.primary_clipboard.connect("owner-change", self.on_primary_owner_change)

		if self.default_text:
			self.history.add(self.default_text)

	def set_text(self, text):
		self.default_clipboard.set_text(text, -1)
		self.primary_clipboard.set_text(text, -1)
		self.default_text = text
		self.primary_text = text
		self.history.add(text)

	def current_text(self):
		return self.default_clipboard.wait_for_text()

	def on_default_owner_change(self, clipboard, event):
		text = clipboard.wait_for_text()
		if text != self.default_text:
			self.default_text = text
			self.history.add(text)

	def on_primary_owner_change(self, clipboard, event):
		text = clipboard.wait_for_text()
		if text != self.primary_text:
			self.primary_text = text
			self.history.add(text, is_from_selection=True)


class ClipboardPopup:
	def __init__(self, history, clipboard):
		self.history = history
		self.clipboard = clipboard
		self.page_rows = 1
		self.filter_text = ""

		self.window = Gtk.Window(title=APP_NAME)
		self.window.set_default_size(420, 520)
		self.window.set_border_width(4)
		self.window.set_skip_taskbar_hint(True)
		self.window.connect("focus-out-event", self.on_focus_out)

		box = Gtk.Box(orientation=Gtk.Orientation.VERTICAL, spacing=4)
		self.window.add(box)

		self.liststore = Gtk.ListStore(str, str, str, int, str)
		self.filter_model = self.liststore.filter_new()
		self.filter_model.set_visible_func(self.visible_func)

		self.treeview = Gtk.TreeView(model=self.filter_model)
		self.treeview.set_headers_visible(False)
		self.treeview.set_enable_search(False)
		self.treeview.connect("button-press-event", self.on_tree_button_press)
		self.treeview.connect("key-press-event", self.on_tree_key_press)
		self.selection = self.treeview.get_selection()

		self.add_text_column(0)
		self.add_text_column(1)
		self.add_text_column(2, weight_column=3)

		scrolled_window = Gtk.ScrolledWindow()
		scrolled_window.set_policy(Gtk.PolicyType.AUTOMATIC, Gtk.PolicyType.AUTOMATIC)
		scrolled_window.add(self.treeview)
		box.pack_start(scrolled_window, True, True, 0)
		self.scrolled_window = scrolled_window

		self.search_entry = Gtk.Entry()
		self.search_entry.set_placeholder_text("Search clipboard history")
		self.search_entry.connect("changed", self.on_search_changed)
		self.search_entry.connect("key-press-event", self.on_entry_key_press)
		box.pack_start(self.search_entry, False, True, 0)

		self.window.get_accessible().set_name(APP_NAME)
		self.treeview.get_accessible().set_name("Clipboard history")
		self.search_entry.get_accessible().set_name("Search clipboard history")

		self.history.connect_changed(lambda items: self.refresh_if_visible())

	def add_text_column(self, text_column, weight_column=None):
		renderer = Gtk.CellRendererText()
		column = Gtk.TreeViewColumn()
		column.pack_start(renderer, True)
		column.add_attribute(renderer, "text", text_column)
		if weight_column is not None:
			column.add_attribute(renderer, "weight", weight_column)
		self.treeview.append_column(column)

	def on_focus_out(self, widget, event):
		widget.hide()
		return False

	def on_tree_button_press(self, widget, event):
		if event.button == 1 and event.type == Gdk.EventType._2BUTTON_PRESS:
			return self.choose_selected()
		return False

	def on_tree_key_press(self, widget, event):
		if event.keyval in (Gdk.KEY_Return, Gdk.KEY_KP_Enter):
			return self.choose_selected()
		if event.keyval in (Gdk.KEY_0, Gdk.KEY_KP_0):
			return self.choose_number(10)
		if Gdk.KEY_1 <= event.keyval <= Gdk.KEY_9:
			return self.choose_number(1 + event.keyval - Gdk.KEY_1)
		if Gdk.KEY_KP_1 <= event.keyval <= Gdk.KEY_KP_9:
			return self.choose_number(1 + event.keyval - Gdk.KEY_KP_1)
		if event.keyval == Gdk.KEY_Escape:
			self.window.hide()
			return True

		unicode_char = Gdk.keyval_to_unicode(event.keyval)
		if unicode_char >= 32:
			self.search_entry.grab_focus()
			self.search_entry.insert_text(chr(unicode_char), -1)
			self.search_entry.set_position(-1)
			return True
		return False

	def on_entry_key_press(self, widget, event):
		if event.keyval in (Gdk.KEY_Return, Gdk.KEY_KP_Enter):
			return self.choose_selected()
		if event.keyval in (Gdk.KEY_Up, Gdk.KEY_KP_Up):
			self.move_selection(-1)
			return True
		if event.keyval in (Gdk.KEY_Down, Gdk.KEY_KP_Down):
			self.move_selection(1)
			return True
		if event.keyval in (Gdk.KEY_Page_Up, Gdk.KEY_KP_Page_Up):
			self.move_selection(-self.page_rows)
			return True
		if event.keyval in (Gdk.KEY_Page_Down, Gdk.KEY_KP_Page_Down):
			self.move_selection(self.page_rows)
			return True
		if event.keyval == Gdk.KEY_Escape:
			self.window.hide()
			return True
		return False

	def on_search_changed(self, entry):
		self.renumber_visible_rows(clear=True)
		self.filter_text = entry.get_text().lower()
		self.filter_model.refilter()
		self.renumber_visible_rows(clear=False)
		self.select_first_row()
		return True

	def visible_func(self, model, store_iter, data=None):
		if self.filter_text == "":
			return True
		clipboard_text = model.get_value(store_iter, 4)
		return self.filter_text in clipboard_text.lower()

	def show(self):
		self.fill_store()
		self.search_entry.set_text("")
		self.select_first_row()
		self.position_near_pointer()
		self.window.show_all()
		self.window.present_with_time(Gtk.get_current_event_time())
		self.treeview.grab_focus()
		self.update_page_rows()

	def refresh_if_visible(self):
		if self.window.get_visible():
			self.fill_store()
			self.select_first_row()

	def fill_store(self):
		current_text = self.clipboard.current_text()
		self.liststore.clear()
		for item_num, item in enumerate(self.history.items, start=1):
			if item_num < 10:
				number = str(item_num)
			elif item_num == 10:
				number = "0"
			else:
				number = ""

			if item == current_text:
				marker = "*"
				font_weight = Pango.Weight.BOLD
			else:
				marker = ""
				font_weight = Pango.Weight.NORMAL

			self.liststore.append([number, marker, format_item(item), font_weight, item])

	def select_first_row(self):
		store_iter = self.filter_model.get_iter_first()
		if store_iter is not None:
			self.treeview.set_cursor(self.filter_model.get_path(store_iter), None, False)

	def choose_selected(self):
		model, store_iter = self.selection.get_selected()
		if store_iter is None:
			return False
		clipboard_text = model.get_value(store_iter, 4)
		self.clipboard.set_text(clipboard_text)
		self.window.hide()
		return True

	def choose_number(self, item_num):
		store_iter = self.filter_model.get_iter_first()
		while item_num > 1 and store_iter is not None:
			item_num -= 1
			store_iter = self.filter_model.iter_next(store_iter)
		if store_iter is None:
			return False
		clipboard_text = self.filter_model.get_value(store_iter, 4)
		self.clipboard.set_text(clipboard_text)
		self.window.hide()
		return True

	def move_selection(self, num_rows):
		model, store_iter = self.selection.get_selected()
		if store_iter is None:
			self.select_first_row()
			return

		if num_rows > 0:
			iter_direction = model.iter_next
		else:
			iter_direction = model.iter_previous
			num_rows = -num_rows

		last_iter = store_iter
		while num_rows > 0 and store_iter is not None:
			num_rows -= 1
			last_iter = store_iter
			store_iter = iter_direction(store_iter)
		if store_iter is None:
			store_iter = last_iter
		self.treeview.set_cursor(model.get_path(store_iter), None, False)

	def renumber_visible_rows(self, clear):
		store_iter = self.filter_model.get_iter_first()
		item_num = 1
		while item_num < 11 and store_iter is not None:
			number = "" if clear else str(item_num if item_num < 10 else 0)
			child_iter = self.filter_model.convert_iter_to_child_iter(store_iter)
			self.liststore.set_value(child_iter, 0, number)
			item_num += 1
			store_iter = self.filter_model.iter_next(store_iter)

	def update_page_rows(self):
		model, store_iter = self.selection.get_selected()
		if store_iter is None:
			self.page_rows = 1
			return
		row_height = self.treeview.get_cell_area(model.get_path(store_iter), None).height
		view_height = self.scrolled_window.get_allocated_height()
		if row_height > 0:
			self.page_rows = max(1, view_height // row_height)

	def position_near_pointer(self):
		display = Gdk.Display.get_default()
		if display is None:
			return
		seat = display.get_default_seat()
		pointer = seat.get_pointer()
		screen, x, y = pointer.get_position()
		monitor = display.get_monitor_at_point(x, y)
		geometry = monitor.get_geometry()
		width, height = self.window.get_size()

		if x + width > geometry.x + geometry.width:
			x = geometry.x + geometry.width - width
		if y + height > geometry.y + geometry.height:
			y = geometry.y + geometry.height - height

		self.window.set_screen(screen)
		self.window.move(max(geometry.x, x), max(geometry.y, y))


class ClipboardDaemon:
	def __init__(self):
		self.history = History()
		self.history.load()
		self.clipboard = ClipboardWatcher(self.history)
		self.popup = ClipboardPopup(self.history, self.clipboard)
		self.server_socket = None

	def run(self):
		GLib.set_application_name(APP_NAME)
		self.setup_socket()
		try:
			Gtk.main()
		except KeyboardInterrupt:
			pass
		finally:
			self.cleanup_socket()

	def setup_socket(self):
		path = socket_path()
		path.parent.mkdir(parents=True, exist_ok=True)
		if path.exists():
			if socket_is_live():
				print("i38-clipboard daemon is already running")
				raise SystemExit(0)
			path.unlink()

		self.server_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
		self.server_socket.setblocking(False)
		self.server_socket.bind(str(path))
		path.chmod(0o600)
		self.server_socket.listen(5)
		GLib.io_add_watch(self.server_socket.fileno(), GLib.IO_IN, self.on_socket_ready)

	def cleanup_socket(self):
		if self.server_socket is not None:
			self.server_socket.close()
			self.server_socket = None
		try:
			socket_path().unlink()
		except OSError:
			pass

	def on_socket_ready(self, _fd, condition):
		if condition & (GLib.IO_ERR | GLib.IO_HUP):
			return False

		try:
			connection, _address = self.server_socket.accept()
		except BlockingIOError:
			return True

		with connection:
			message = connection.recv(1024).strip()
			if message == SOCKET_MESSAGE_SHOW.strip():
				GLib.idle_add(self.popup.show)
				connection.sendall(b"ok\n")
			elif message == SOCKET_MESSAGE_PING.strip():
				connection.sendall(b"ok\n")
			else:
				connection.sendall(b"unknown command\n")
		return True


def send_request(message):
	try:
		with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client:
			client.settimeout(1)
			client.connect(str(socket_path()))
			client.sendall(message)
			response = client.recv(1024)
		return response.startswith(b"ok")
	except OSError:
		return False


def send_show_request():
	return send_request(SOCKET_MESSAGE_SHOW)


def socket_is_live():
	return send_request(SOCKET_MESSAGE_PING)


def parse_args():
	parser = argparse.ArgumentParser(description=APP_NAME)
	mode = parser.add_mutually_exclusive_group()
	mode.add_argument("--daemon", action="store_true", help="run the clipboard history daemon")
	mode.add_argument("--show", action="store_true", help="show clipboard history from the running daemon")
	return parser.parse_args()


def main():
	args = parse_args()
	if args.show:
		if send_show_request():
			return 0
		print("i38-clipboard daemon is not running", file=sys.stderr)
		return 1

	if not Gtk.init_check()[0]:
		print("i38-clipboard could not connect to a GTK display", file=sys.stderr)
		return 1

	return ClipboardDaemon().run() or 0


if __name__ == "__main__":
	raise SystemExit(main())