added clipcascade

This commit is contained in:
liph
2026-07-23 16:48:58 +02:00
parent 90e44adbae
commit cba86d9dce
31 changed files with 6406 additions and 0 deletions
+464
View File
@@ -0,0 +1,464 @@
import base64
import io
import json
import logging
import os
import xxhash
from PIL import Image
from core.constants import *
from core.config import Config
if PLATFORM.startswith(LINUX) and LINUX_USE_CLI_UI:
from cli.tray import TaskbarPanel
else:
from gui.tray import TaskbarPanel
if PLATFORM == WINDOWS or PLATFORM == MACOS:
import pyperclip
if PLATFORM == WINDOWS:
import win32clipboard
from clipboard import clipboard_monitor_win as clipboard_monitor
elif PLATFORM == MACOS:
import pasteboard
from clipboard import clipboard_monitor_mac as clipboard_monitor
elif PLATFORM.startswith(LINUX):
from clipboard import clipboard_monitor_linux as clipboard_monitor
import subprocess
class ClipboardManager:
def __init__(self, config: Config):
self.config = config
self.previous_clipboard_hash = 0
self.sys_tray: TaskbarPanel = None
self.is_files_download_enabled = False
if PLATFORM.startswith(LINUX) and XMODE:
self.is_x_clipboard_owner = clipboard_monitor.is_x_clipboard_owner()
def set_tray_ref(self, sys_tray: TaskbarPanel):
"""
Sets the system tray reference.
"""
self.sys_tray = sys_tray
def reset_files_download(self):
"""
Resets the files download flag and disables the download functionality in the system tray if enabled.
"""
if self.is_files_download_enabled:
self.is_files_download_enabled = False
if self.sys_tray:
self.sys_tray.disable_files_download()
@staticmethod
def hash_clipboard(clipboard: str) -> int:
return xxhash.xxh64(clipboard).intdigest()
def is_clipboard_size_within_limit(
self, clipboard_content: any, type_: str = "text"
) -> bool:
if clipboard_content is None:
raise ValueError("Clipboard content cannot be None")
content_size_in_bytes = None
if type_ == "text":
content_size_in_bytes = len(clipboard_content.encode("utf-8"))
elif type_ == "image":
content_size_in_bytes = ClipboardManager.get_image_size(
img=clipboard_content
)
elif type_ == "files":
content_size_in_bytes = ClipboardManager.calculate_cumulative_file_size(
files=clipboard_content
)
# Check if the content size exceeds the server limit
max_allowed_size = self.config.data["maxsize"]
if (
max_allowed_size is not None
and max_allowed_size >= 0
and content_size_in_bytes > max_allowed_size
):
logging.warning(
"Clipboard content size exceeds the maximum allowed limit. "
f"Allowed: {max_allowed_size} bytes, Found: {content_size_in_bytes} bytes."
)
return False
# Check if the content size exceeds the local limit
local_clipboard_size_limit = self.config.data[
"max_clipboard_size_local_limit_bytes"
]
if local_clipboard_size_limit is not None and local_clipboard_size_limit >= 0:
if content_size_in_bytes > local_clipboard_size_limit:
logging.warning(
"Clipboard content size exceeds the local allowed limit. "
f"Allowed: {local_clipboard_size_limit} bytes, Found: {content_size_in_bytes} bytes."
)
return False
return True
def has_clipboard_changed(self, payload: str) -> bool:
"""
Check if the clipboard content has changed by comparing the current hash
with the previous clipboard hash.
Parameters:
- payload: The current clipboard content.
Returns:
- True if the clipboard content has changed, False otherwise.
"""
current_clipboard_hash = ClipboardManager.hash_clipboard(payload)
if current_clipboard_hash != self.previous_clipboard_hash:
self.previous_clipboard_hash = current_clipboard_hash
return True
return False
def on_copy(self, copy_callback):
clipboard_monitor.on_update(
callback=lambda type_, content: self.clipboard_to_base64(
copy_callback, content, type_
),
enable_image_monitoring=self.config.data["enable_image_sharing"],
enable_file_monitoring=self.config.data["enable_file_sharing"],
)
def clipboard_to_base64(self, callback, content: any, type_: str = "text"):
try:
self.reset_files_download()
type_ = type_.lower()
if type_ == "text":
if self.is_clipboard_size_within_limit(content, type_):
callback(content, type_)
elif type_ == "image":
if isinstance(content, list):
if content is None or len(content) == 0:
raise ValueError(
"Clipboard image content cannot be None or empty"
)
content = Image.open(content[0])
if self.is_clipboard_size_within_limit(content, type_):
content_str = ClipboardManager.convert_image_to_base64(img=content)
callback(content_str, type_)
elif type_ == "files":
if PLATFORM.startswith(LINUX):
if content is not None and len(content) > 0:
temp = []
for path in content:
if path.startswith("file:"):
temp.append(path[5:])
else:
temp.append(path)
content = temp
if self.is_clipboard_size_within_limit(content, type_):
content_str = ClipboardManager.convert_files_to_base64(
file_paths=content
)
if content_str != "{}": # Check if the JSON string is empty
callback(content_str, type_)
except Exception as e:
logging.error(f"Failed to convert clipboard data to base64: {e}")
def base64_to_clipboard(self, base64_string: str, type_: str = "text"):
try:
if type_ == "text":
txt = base64_string
if self.is_clipboard_size_within_limit(txt, type_):
self.paste(txt, type_)
elif type_ == "image":
img = ClipboardManager.convert_base64_to_image(base64_img=base64_string)
if self.is_clipboard_size_within_limit(img, type_):
self.paste(img, type_)
elif type_ == "files":
file_objects = ClipboardManager.convert_base64_to_files(
base64_json=base64_string
)
if self.is_clipboard_size_within_limit(file_objects, type_):
self.paste(file_objects, type_)
except Exception as e:
logging.error(f"Failed to convert base64 data to clipboard: {e}")
@staticmethod
def execute_command(*args, input_data):
"""
Execute a command with input data.
Parameters:
- *args: Positional arguments for the command.
- input_data: Input data to be passed to the command.
"""
if PLATFORM.startswith(LINUX):
try:
subprocess.run(
args,
input=input_data,
check=True,
)
except Exception as e:
logging.error(f"Failed to execute command: {e}")
raise
def paste(self, payload: any, payload_type: str = "text"):
try:
self.reset_files_download()
if payload_type == "text":
if PLATFORM == WINDOWS or PLATFORM == MACOS:
pyperclip.copy(payload)
elif PLATFORM.startswith(LINUX):
if XMODE and self.is_x_clipboard_owner:
ClipboardManager.execute_command(
"xclip",
"-selection",
"clipboard",
input_data=payload.encode("utf-8"),
)
else:
ClipboardManager.execute_command(
"wl-copy",
input_data=payload.encode("utf-8"),
)
elif payload_type == "image":
if PLATFORM == WINDOWS:
# Save the image to a binary buffer in BMP format
with io.BytesIO() as output:
payload.convert("RGB").save(output, format="BMP")
bmp_data = output.getvalue()[14:] # Skip BMP header (14 bytes)
win32clipboard.OpenClipboard()
win32clipboard.EmptyClipboard()
clipboard_monitor.enable_block_image_once() # Block image copy to prevent deadlock
win32clipboard.SetClipboardData(win32clipboard.CF_DIB, bmp_data)
win32clipboard.CloseClipboard()
elif PLATFORM == MACOS:
# Save the image to a binary buffer in TIFF format
with io.BytesIO() as output:
payload.convert("RGB").save(output, format="TIFF")
tiff_data = output.getvalue()
clipboard_monitor.enable_block_image_once() # Block image copy to prevent deadlock
clipboard_monitor.write_to_pasteboard(tiff_data, pasteboard.TIFF)
elif PLATFORM.startswith(LINUX):
# Save the image to a binary buffer in PNG format
with io.BytesIO() as output:
payload.convert("RGB").save(output, format="PNG")
png_data = output.getvalue()
clipboard_monitor.enable_block_image_once() # Block image copy to prevent deadlock
if XMODE and self.is_x_clipboard_owner:
ClipboardManager.execute_command(
"xclip",
"-selection",
"clipboard",
"-t",
"image/png",
input_data=png_data,
)
else:
ClipboardManager.execute_command(
"wl-copy",
"--type",
"image/png",
input_data=png_data,
)
elif payload_type == "files":
if (
payload is not None
and isinstance(payload, dict)
and len(payload) > 0
):
if self.sys_tray is not None:
self.is_files_download_enabled = True
self.sys_tray.enable_files_download(files=payload)
except Exception as e:
logging.error(f"Failed to copy data to clipboard: {e}")
raise
def stop(self):
self.reset_files_download()
clipboard_monitor.stop()
@staticmethod
def calculate_cumulative_file_size(files: tuple | list | dict) -> int:
"""
Calculate the cumulative size of a list of files.
Args:
files (tuple or list): A tuple of file paths.
files (dict): A dictionary of files with file names as keys and file object as values.
Returns:
int: The cumulative size of the files in bytes.
Raises:
IOError: If a file cannot be read or processed.
"""
cumulative_size = 0
if isinstance(files, tuple | list):
for file_path in files:
try:
if os.path.isfile(file_path):
cumulative_size += os.path.getsize(file_path)
except Exception as e:
raise IOError(
f"Failed to calculate size for file '{file_path}' {e}."
) from e
if isinstance(files, dict):
for file_name, file_object in files.items():
try:
cumulative_size += file_object.getbuffer().nbytes
except Exception as e:
raise IOError(
f"Failed to calculate size for file '{file_name}' {e}."
) from e
return cumulative_size
@staticmethod
def convert_files_to_base64(file_paths: tuple | list) -> str:
"""
Converts a list of files to base64-encoded strings and returns them as a JSON string.
Args:
file_paths (tuple or list): A tuple of file paths to be converted.
Returns:
str: A JSON string where file names are keys and base64-encoded content is values.
Raises:
IOError: If a file cannot be read or processed.
"""
base64_encoded_files = {}
for file_path in file_paths:
try:
if os.path.isfile(file_path):
file_name = os.path.basename(file_path)
with open(file_path, "rb") as file:
encoded_string = base64.b64encode(file.read()).decode("utf-8")
base64_encoded_files[file_name] = encoded_string
except Exception as e:
raise IOError(f"Failed to process file '{file_path}'. {e}") from e
return json.dumps(base64_encoded_files)
@staticmethod
def convert_base64_to_files(base64_json: dict) -> dict:
"""
Converts a JSON string with base64-encoded file content to a dictionary of file-like objects.
Args:
base64_json (str): A JSON string where file names are keys and base64-encoded content is values.
Returns:
dict: A dictionary where keys are file names, and values are BytesIO objects containing the decoded file data.
"""
file_objects = {}
try:
base64_data = json.loads(base64_json)
for file_name, encoded_content in base64_data.items():
decoded_content = base64.b64decode(encoded_content)
file_objects[file_name] = io.BytesIO(decoded_content)
except Exception as e:
raise IOError(f"Error processing base64 JSON. {e}") from e
return file_objects
@staticmethod
def get_image_size(img: Image.Image | bytes) -> int:
"""
Calculate the size of an image in bytes.
Args:
img (Image.Image | bytes): The image to calculate the size for.
Returns:
int: The size of the image in bytes.
Raises:
IOError: If the image cannot be processed or saved.
"""
try:
if isinstance(img, bytes):
size_in_bytes = len(img)
if isinstance(img, Image.Image):
with io.BytesIO() as buffer:
img.save(buffer, format=img.format or "PNG")
size_in_bytes = buffer.tell()
return size_in_bytes
except Exception as e:
raise IOError(f"Failed to calculate the image size. {e}") from e
@staticmethod
def convert_image_to_base64(img: Image.Image | bytes) -> str:
"""
Converts an image to a base64-encoded string.
Args:
img (Image.Image | bytes): The image to be converted.
Returns:
str: The base64-encoded string representation of the image.
Raises:
IOError: If the image cannot be processed or saved.
"""
try:
if isinstance(img, bytes):
base64_string = base64.b64encode(img).decode("utf-8")
if isinstance(img, Image.Image):
with io.BytesIO() as buffered:
img.save(buffered, format=img.format or "PNG")
base64_string = base64.b64encode(buffered.getvalue()).decode(
"utf-8"
)
return base64_string
except Exception as e:
raise IOError(f"Failed to convert the image to base64. {e}") from e
@staticmethod
def convert_base64_to_image(base64_img: str) -> Image.Image:
"""
Converts a base64-encoded string to a PIL Image object.
Args:
base64_img (str): The base64-encoded string to be converted.
Returns:
Image.Image: A PIL Image object representing the decoded image.
Raises:
ValueError: If the base64 string is invalid.
IOError: If the image cannot be processed.
"""
try:
# Decode the base64 string
image_data = base64.b64decode(base64_img)
except base64.binascii.Error as e:
raise ValueError("Invalid base64 string.") from e
try:
# Load the image from the decoded bytes
with io.BytesIO(image_data) as image_stream:
image = Image.open(image_stream)
image.load()
return image
except IOError as e:
raise IOError(f"Failed to process the image. {e}") from e
+486
View File
@@ -0,0 +1,486 @@
import logging
import re
import subprocess
import threading
import time
from core.constants import *
_callback_update = None
_clipboard_thread = None
_block_image_once = False
_is_gdk_running = False
_run_poll = threading.Event()
_wl_watch_proc = None
def _on_clipboard_changed(
clipboard, event=None, enable_image_monitoring=False, enable_file_monitoring=False
):
global _block_image_once
# Files
if enable_file_monitoring:
uris = clipboard.wait_for_uris()
if uris is not None and len(uris) > 0:
if _callback_update:
_callback_update("files", uris)
return
# Text
text = clipboard.wait_for_text()
if text is not None and len(text) > 0:
if _callback_update:
_callback_update("text", text)
return
# Image
if enable_image_monitoring:
pixbuf = clipboard.wait_for_image()
if pixbuf is not None:
if _block_image_once:
_block_image_once = False
return
if _callback_update:
success, buffer = pixbuf.save_to_bufferv("png")
if success:
_callback_update("image", bytes(buffer))
else:
logging.error("Failed to convert image(pixbuf) to buffer")
return
def _monitor_x_wl_clipboard(
x_mode: bool,
enable_image_monitoring=False,
enable_file_monitoring=False,
):
global _block_image_once
last_error = None
previous_clipboard: str | bytes | None = None
ignore_patterns = [
r"target .+ not available", # xclip pattern
r"no suitable type of content copied", # wl-clipboard pattern
]
if LINUX_CLIPBOARD_POLL_INTERVAL_SEC is not None:
timeout = LINUX_CLIPBOARD_POLL_INTERVAL_SEC
logging.info(f"Clipboard polling interval (--polling): {timeout}s")
elif x_mode:
timeout = 0.3 # xclip seconds
else:
timeout = 3 # wl-clipboard seconds
while _run_poll.is_set():
if x_mode:
success, mime_list = execute_command(
"xclip", "-selection", "clipboard", "-t", "TARGETS", "-o"
)
else:
success, mime_list = execute_command("wl-paste", "-l")
if not success:
error_msg = f"Failed to retrieve MIME types: {mime_list}"
if error_msg != last_error:
logging.error(error_msg)
last_error = error_msg
time.sleep(timeout)
continue
mime_list = mime_list.decode("utf-8")
mime_list = mime_list.replace("\r\n", "\n").replace("\r", "\n").split("\n")
mime_list = [m.strip() for m in mime_list if len(m.strip()) > 0]
type_ = convert_mime_to_generic_type(mime_list)
# Text
if type_ == "text":
if x_mode:
success, text = execute_command(
"xclip", "-selection", "clipboard", "-o"
)
else:
success, text = execute_command("wl-paste", "-n")
if success:
text = text.decode("utf-8")
if len(text) > 0 and text != previous_clipboard:
previous_clipboard = text
if _callback_update:
_callback_update("text", text)
else:
error_msg = f"Failed to retrieve text content from clipboard. {text}"
if error_msg != last_error:
if not any(
re.search(pattern, error_msg.lower())
for pattern in ignore_patterns
):
logging.error(error_msg)
last_error = error_msg
# Image
if type_ == "image" and enable_image_monitoring:
if x_mode:
success, image = execute_command(
"xclip",
"-selection",
"clipboard",
"-t",
"image/png",
"-o",
)
else:
success, image = execute_command("wl-paste", "-t", "image/png")
if success:
if image != previous_clipboard:
previous_clipboard = image
if _callback_update and not _block_image_once:
_callback_update("image", image)
else:
_block_image_once = False
else:
error_msg = f"Failed to retrieve image content from clipboard. {image}"
if error_msg != last_error:
if not any(
re.search(pattern, error_msg.lower())
for pattern in ignore_patterns
):
logging.error(error_msg)
last_error = error_msg
# Files
if type_ == "files" and enable_file_monitoring:
if x_mode:
success, files = execute_command(
"xclip",
"-selection",
"clipboard",
"-t",
"text/uri-list",
"-o",
)
else:
success, files = execute_command(
"wl-paste", "-t", "text/uri-list", "-n"
)
if success:
files = files.decode("utf-8")
files = files.replace("\r\n", "\n").replace("\r", "\n").split("\n")
files = [f.strip() for f in files if len(f.strip()) > 0]
if files != previous_clipboard:
previous_clipboard = files
if _callback_update:
_callback_update("files", files)
else:
error_msg = f"Failed to retrieve files content from clipboard. {files}"
if error_msg != last_error:
if not any(
re.search(pattern, error_msg.lower())
for pattern in ignore_patterns
):
logging.error(error_msg)
last_error = error_msg
time.sleep(timeout)
def _monitor_wl_watch(enable_image_monitoring=False, enable_file_monitoring=False):
"""Event-driven Wayland clipboard monitoring using wl-paste --watch.
Uses the wlr-data-control-v1 protocol which does not create visible
surfaces or steal focus. Supported by wlroots-based compositors
(Sway, Hyprland, etc.) and KDE Plasma on Wayland.
Returns True if watch mode ran successfully, False to fall back to polling."""
global _block_image_once, _wl_watch_proc
last_error = None
previous_clipboard = None
ignore_patterns = [
r"target .+ not available",
r"no suitable type of content copied",
]
try:
_wl_watch_proc = subprocess.Popen(
["wl-paste", "--watch", "echo"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
time.sleep(0.5)
if _wl_watch_proc.poll() is not None:
stderr_out = _wl_watch_proc.stderr.read().decode("utf-8", errors="ignore")
logging.warning(
f"wl-paste --watch exited immediately: {stderr_out.strip()}"
)
_wl_watch_proc = None
return False
logging.info(
"Using wl-paste --watch for clipboard monitoring (no focus stealing)"
)
while _run_poll.is_set():
line = _wl_watch_proc.stdout.readline()
if not line:
break
if not _run_poll.is_set():
break
success, mime_output = execute_command("wl-paste", "-l")
if not success:
error_msg = f"Failed to retrieve MIME types: {mime_output}"
if error_msg != last_error:
logging.error(error_msg)
last_error = error_msg
continue
mime_list = mime_output.decode("utf-8")
mime_list = mime_list.replace("\r\n", "\n").replace("\r", "\n").split("\n")
mime_list = [m.strip() for m in mime_list if len(m.strip()) > 0]
type_ = convert_mime_to_generic_type(mime_list)
# Text
if type_ == "text":
success, text = execute_command("wl-paste", "-n")
if success:
text = text.decode("utf-8")
if len(text) > 0 and text != previous_clipboard:
previous_clipboard = text
if _callback_update:
_callback_update("text", text)
else:
error_msg = (
f"Failed to retrieve text content from clipboard. {text}"
)
if error_msg != last_error:
if not any(
re.search(pattern, error_msg.lower())
for pattern in ignore_patterns
):
logging.error(error_msg)
last_error = error_msg
# Image
elif type_ == "image" and enable_image_monitoring:
success, image = execute_command("wl-paste", "-t", "image/png")
if success:
if image != previous_clipboard:
previous_clipboard = image
if _callback_update and not _block_image_once:
_callback_update("image", image)
else:
_block_image_once = False
else:
error_msg = (
f"Failed to retrieve image content from clipboard. {image}"
)
if error_msg != last_error:
if not any(
re.search(pattern, error_msg.lower())
for pattern in ignore_patterns
):
logging.error(error_msg)
last_error = error_msg
# Files
elif type_ == "files" and enable_file_monitoring:
success, files = execute_command(
"wl-paste", "-t", "text/uri-list", "-n"
)
if success:
files = files.decode("utf-8")
files = (
files.replace("\r\n", "\n").replace("\r", "\n").split("\n")
)
files = [f.strip() for f in files if len(f.strip()) > 0]
if files != previous_clipboard:
previous_clipboard = files
if _callback_update:
_callback_update("files", files)
else:
error_msg = (
f"Failed to retrieve files content from clipboard. {files}"
)
if error_msg != last_error:
if not any(
re.search(pattern, error_msg.lower())
for pattern in ignore_patterns
):
logging.error(error_msg)
last_error = error_msg
return True
except FileNotFoundError:
logging.warning("wl-paste not found, cannot use --watch mode")
return False
except Exception as e:
logging.warning(f"wl-paste --watch failed: {e}")
return False
finally:
if _wl_watch_proc is not None:
_wl_watch_proc.terminate()
try:
_wl_watch_proc.wait(timeout=2)
except subprocess.TimeoutExpired:
_wl_watch_proc.kill()
_wl_watch_proc = None
def convert_mime_to_generic_type(mime_list):
if "text/uri-list" in mime_list:
return "files"
if any(mime.startswith("image/") for mime in mime_list):
return "image"
text_mime = [
"text/plain",
"text/plain;charset=utf-8",
"STRING",
"TEXT",
"COMPOUND_TEXT",
"UTF8_STRING",
]
if any(t_mime in mime_list for t_mime in text_mime):
return "text"
return "unknown"
def execute_command(*args) -> tuple:
"""
Executes a command with the given arguments and returns the output or error.
Parameters:
*args: Variable-length argument list to be passed as the command and arguments.
Returns:
tuple: (success: bool, result: str)
success is True if the command executed successfully, False otherwise.
result is the output or error of the command.
"""
process = subprocess.Popen(
args,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
output, error = process.communicate()
if process.returncode == 0: # Success
return (True, output)
else: # Failure
return (False, error.decode())
def is_x_clipboard_owner():
# Check if the X clipboard is owned by the current user
return execute_command("xclip", "-selection", "clipboard", "-t", "TARGETS", "-o")[0]
def _start_clipboard_polling(enable_image_monitoring, enable_file_monitoring):
if XMODE:
x_clipboard_owner = is_x_clipboard_owner()
if not x_clipboard_owner:
logging.warning(
"x-clip is not owned by the current user. Switching to wl-clipboard."
)
_monitor_x_wl_clipboard(
x_mode=x_clipboard_owner,
enable_image_monitoring=enable_image_monitoring,
enable_file_monitoring=enable_file_monitoring,
)
else:
if not _monitor_wl_watch(
enable_image_monitoring=enable_image_monitoring,
enable_file_monitoring=enable_file_monitoring,
):
logging.info(
"Falling back to wl-paste polling mode for clipboard monitoring"
)
_monitor_x_wl_clipboard(
x_mode=False,
enable_image_monitoring=enable_image_monitoring,
enable_file_monitoring=enable_file_monitoring,
)
def _runner(enable_image_monitoring=False, enable_file_monitoring=False):
global _is_gdk_running, _run_poll
logging.info(f"XMODE: {XMODE}")
try:
_run_poll.set()
import gi
gi.require_version("Gtk", "3.0")
gi.require_version("Gdk", "3.0")
from gi.repository import Gtk, Gdk
if "x11" in str(type(Gdk.Display.get_default())).lower(): # X11
logging.info("Starting GTK clipboard monitoring for X11 display server.")
clipboard = Gtk.Clipboard.get(Gdk.SELECTION_CLIPBOARD)
clipboard.connect(
"owner-change",
lambda clip, event: _on_clipboard_changed(
clip, event, enable_image_monitoring, enable_file_monitoring
),
)
_is_gdk_running = True
Gtk.main()
else:
logging.warning(
f"Unsupported display server detected ${str(type(Gdk.Display.get_default())).lower()}. Starting polling mode for {detect_linux_display_server()} server as fallback."
)
_start_clipboard_polling(enable_image_monitoring, enable_file_monitoring)
except Exception as e:
logging.error(
f"Failed to start clipboard monitor: Error {e}\nStarting polling mode for {detect_linux_display_server()} server as fallback."
)
_start_clipboard_polling(enable_image_monitoring, enable_file_monitoring)
def _start(enable_image_monitoring=False, enable_file_monitoring=False):
global _clipboard_thread
if not _clipboard_thread:
_clipboard_thread = threading.Thread(
target=_runner,
args=(enable_image_monitoring, enable_file_monitoring),
daemon=True,
)
_clipboard_thread.start()
def stop():
global _clipboard_thread, _callback_update, _block_image_once, _run_poll, _is_gdk_running, _wl_watch_proc
if _clipboard_thread:
if _is_gdk_running:
import gi
gi.require_version("Gtk", "3.0")
from gi.repository import Gtk
Gtk.main_quit()
_is_gdk_running = False
_run_poll.clear()
if _wl_watch_proc is not None:
_wl_watch_proc.terminate()
_clipboard_thread.join() # Wait for the thread to finish
_clipboard_thread = None
_callback_update = None
_block_image_once = False
_wl_watch_proc = None
logging.info("Clipboard monitor stopped")
def wait():
global _clipboard_thread
if _clipboard_thread:
_clipboard_thread.join()
def enable_block_image_once():
global _block_image_once
_block_image_once = True
def on_update(callback, enable_image_monitoring=False, enable_file_monitoring=False):
global _callback_update
_callback_update = callback
_start(enable_image_monitoring, enable_file_monitoring)
+149
View File
@@ -0,0 +1,149 @@
from io import BytesIO
import logging
import threading
import time
import pasteboard
_clipboard_thread = None
_callback_update = None
_run = False
_first_run = False
_block_image_once = False
_pasteboard_lock = threading.Lock()
_pb_writer = None
def write_to_pasteboard(data, pb_type):
global _pb_writer
with _pasteboard_lock:
if _pb_writer is None:
_pb_writer = pasteboard.Pasteboard()
_pb_writer.set_contents(data, pb_type)
def _runner(enable_image_monitoring=False, enable_file_monitoring=False):
global _first_run, _block_image_once
try:
pb_text = pasteboard.Pasteboard()
if enable_image_monitoring:
pb_image_png = pasteboard.Pasteboard()
pb_image_tiff = pasteboard.Pasteboard()
if enable_file_monitoring:
pb_files = pasteboard.Pasteboard()
image_processed = False
files_processed = False
while _run:
# don't change the execution order (files,text,image or files,image,text)
if enable_file_monitoring:
# Files
with _pasteboard_lock:
clipboard_files = pb_files.get_file_urls(diff=True)
if (
clipboard_files is not None
and type(clipboard_files) is tuple
and len(clipboard_files) > 0
):
if _callback_update and not _first_run:
files_processed = True
_callback_update("files", clipboard_files)
# Text
with _pasteboard_lock:
clipboard_text = pb_text.get_contents(
type=pasteboard.String, diff=True
) # If True, retrieves and returns the content only if it has changed since the last call.
# This approach is efficient even in cases of frequent polling.
if (
clipboard_text is not None
and type(clipboard_text) is str
and len(clipboard_text) > 0
):
if _callback_update and not _first_run and not files_processed:
_callback_update("text", clipboard_text)
if enable_image_monitoring:
# Image (PNG)
with _pasteboard_lock:
clipboard_image_png = pb_image_png.get_contents(
type=pasteboard.PNG, diff=True
)
if (
clipboard_image_png is not None
and type(clipboard_image_png) is bytes
):
if _callback_update and not _first_run and not image_processed:
image_processed = True
if _block_image_once:
_block_image_once = False
else:
if not files_processed:
_callback_update("image", clipboard_image_png)
# Image (TIFF)
with _pasteboard_lock:
clipboard_image_tiff = pb_image_tiff.get_contents(
type=pasteboard.TIFF, diff=True
)
if (
clipboard_image_tiff is not None
and type(clipboard_image_tiff) is bytes
):
if _callback_update and not _first_run and not image_processed:
image_processed = True
if _block_image_once:
_block_image_once = False
else:
if not files_processed:
_callback_update("image", clipboard_image_tiff)
files_processed = False
image_processed = False
_first_run = False
time.sleep(0.3) # seconds
except Exception as e:
logging.error(f"Error processing clipboard update: {e}")
def _start(enable_image_monitoring=False, enable_file_monitoring=False):
global _clipboard_thread, _run, _first_run
if not _clipboard_thread:
_run = True
_first_run = True
_clipboard_thread = threading.Thread(
target=_runner,
args=(enable_image_monitoring, enable_file_monitoring),
daemon=True,
)
_clipboard_thread.start()
def stop():
global _clipboard_thread, _callback_update, _run, _first_run, _block_image_once, _pb_writer
if _clipboard_thread:
_run = False
_clipboard_thread.join() # Wait for the thread to finish
_first_run = False
_clipboard_thread = None
_callback_update = None
_block_image_once = False
_pb_writer = None
logging.info("Clipboard monitor stopped")
def wait():
global _clipboard_thread
if _clipboard_thread:
_clipboard_thread.join()
def enable_block_image_once():
global _block_image_once
_block_image_once = True
def on_update(callback, enable_image_monitoring=False, enable_file_monitoring=False):
global _callback_update
_callback_update = callback
_start(enable_image_monitoring, enable_file_monitoring)
+163
View File
@@ -0,0 +1,163 @@
import logging
import win32gui
import win32api
import win32con
import win32clipboard
import threading
import ctypes
import time
from PIL import ImageGrab
_clipboard_thread = None
_hwnd = None # Store the window handle
_callback_update = None
_block_image_once = False
def _get_clipboard_content(enable_image_monitoring=False, enable_file_monitoring=False):
"""
Get the content of the clipboard.
Image:
PNG -> PngImagePlugin.PngImageFile
DIB -> BmpImagePlugin.DibImageFile
PNG, DIB, JPG, etc. -> [file_path1, file_path2, ...]
Text:
CF_UNICODETEXT, CF_TEXT -> str
Files:
CF_HDROP -> (file_path1, file_path2, ...)
"""
# sleep 0.5 to avoid clipboard not ready for read
time.sleep(0.5)
clipboard_type = None
clipboard_content = None
if enable_image_monitoring and win32clipboard.IsClipboardFormatAvailable(
win32con.CF_BITMAP
):
clipboard_type = "image"
clipboard_content = ImageGrab.grabclipboard()
else:
win32clipboard.OpenClipboard()
try:
if win32clipboard.IsClipboardFormatAvailable(win32con.CF_UNICODETEXT):
text = win32clipboard.GetClipboardData(win32con.CF_UNICODETEXT)
clipboard_type = "text"
clipboard_content = text
elif win32clipboard.IsClipboardFormatAvailable(win32con.CF_TEXT):
text_bytes = win32clipboard.GetClipboardData(win32con.CF_TEXT)
text = text_bytes.decode()
clipboard_type = "text"
clipboard_content = text
elif enable_file_monitoring and win32clipboard.IsClipboardFormatAvailable(
win32con.CF_HDROP
):
files = win32clipboard.GetClipboardData(win32con.CF_HDROP)
clipboard_type = "files"
clipboard_content = files
finally:
win32clipboard.CloseClipboard()
return (clipboard_type, clipboard_content)
def _process_message(
hwnd: int,
msg: int,
wparam: int,
lparam: int,
enable_image_monitoring=False,
enable_file_monitoring=False,
):
global _block_image_once
WM_CLIPBOARDUPDATE = 0x031D
if msg == WM_CLIPBOARDUPDATE:
clip = _get_clipboard_content(enable_image_monitoring, enable_file_monitoring)
try:
if clip[0] == "text" and _callback_update:
_callback_update(clip[0], clip[1])
if enable_image_monitoring and clip[0] == "image" and _callback_update:
if _block_image_once:
_block_image_once = False
else:
_callback_update(clip[0], clip[1])
if enable_file_monitoring and clip[0] == "files" and _callback_update:
_callback_update(clip[0], clip[1])
except Exception as e:
logging.error(f"Error processing clipboard update: {e}")
return 0
def _create_window(enable_image_monitoring=False, enable_file_monitoring=False):
global _hwnd
className = "ClipboardHook"
wc = win32gui.WNDCLASS()
wc.lpfnWndProc = lambda hwnd, msg, wparam, lparam: _process_message(
hwnd, msg, wparam, lparam, enable_image_monitoring, enable_file_monitoring
)
wc.lpszClassName = className
wc.hInstance = win32api.GetModuleHandle(None)
class_atom = win32gui.RegisterClass(wc)
_hwnd = win32gui.CreateWindow(
class_atom, className, 0, 0, 0, 0, 0, 0, 0, wc.hInstance, None
)
def _runner(enable_image_monitoring=False, enable_file_monitoring=False):
global _hwnd
_create_window(enable_image_monitoring, enable_file_monitoring)
ctypes.windll.user32.AddClipboardFormatListener(_hwnd)
try:
win32gui.PumpMessages()
finally:
ctypes.windll.user32.RemoveClipboardFormatListener(_hwnd)
win32gui.DestroyWindow(_hwnd)
win32gui.UnregisterClass("ClipboardHook", win32api.GetModuleHandle(None))
def _start(enable_image_monitoring=False, enable_file_monitoring=False):
global _clipboard_thread
if not _clipboard_thread:
_clipboard_thread = threading.Thread(
target=_runner,
args=(enable_image_monitoring, enable_file_monitoring),
daemon=True,
)
_clipboard_thread.start()
def stop():
global _clipboard_thread, _hwnd, _callback_update, _block_image_once
if _clipboard_thread and _hwnd:
win32gui.PostMessage(
_hwnd, win32con.WM_QUIT, 0, 0
) # Send WM_QUIT to the window
_clipboard_thread.join() # Wait for the thread to finish
_clipboard_thread = None
_hwnd = None
_callback_update = None
_block_image_once = False
logging.info("Clipboard monitor stopped")
def wait():
global _clipboard_thread
if _clipboard_thread:
_clipboard_thread.join()
def enable_block_image_once():
global _block_image_once
_block_image_once = True
def on_update(callback, enable_image_monitoring=False, enable_file_monitoring=False):
global _callback_update
_callback_update = callback
_start(enable_image_monitoring, enable_file_monitoring)