1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
|
"""
The system clipboard, for the chat's right-click copy and paste.
A copy goes to the terminal (OSC 52, through Textual), which reaches the
clipboard of the machine in front of the person even over SSH, and to the
platform's own tool too, for terminals that ignore OSC 52. A paste has no
terminal escape to rely on: the text is read with the platform's tool, else it
is what this program copied last. A tool that is missing, fails or hangs is
skipped, never an error.
"""
import asyncio
import os
import shutil
import subprocess
import sys
TIMEOUT = 3.0
_PS = ["powershell", "-NoProfile", "-NonInteractive", "-Command"]
def _tools(read: bool) -> list[list[str]]:
if sys.platform == "win32":
return [_PS + ["[Console]::OutputEncoding = [Text.Encoding]::UTF8; Get-Clipboard -Raw"]
if read else ["clip"]]
if sys.platform == "darwin":
return [["pbpaste"] if read else ["pbcopy"]]
tools = []
if os.environ.get("WAYLAND_DISPLAY"):
tools.append(["wl-paste", "--no-newline"] if read else ["wl-copy"])
if os.environ.get("DISPLAY"):
tools.append(["xclip", "-selection", "clipboard", "-o"] if read
else ["xclip", "-selection", "clipboard", "-i"])
tools.append(["xsel", "--clipboard", "--output"] if read
else ["xsel", "--clipboard", "--input"])
return tools
def _encode(text: str) -> bytes:
# clip.exe reads the console's code page unless the text says it is UTF-16.
return ("" + text).encode("utf-16-le") if sys.platform == "win32" else text.encode()
def _run_blocking(cmd: list[str], data: bytes | None) -> bytes | None:
if shutil.which(cmd[0]) is None:
return None
env = {**os.environ, "LC_CTYPE": "UTF-8"} if sys.platform == "darwin" else None
try:
proc = subprocess.run(
cmd, input=data, stdin=None if data is not None else subprocess.DEVNULL,
stdout=subprocess.PIPE if data is None else subprocess.DEVNULL,
stderr=subprocess.DEVNULL, env=env, timeout=TIMEOUT,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
except (OSError, subprocess.TimeoutExpired):
return None
return (proc.stdout or b"") if proc.returncode == 0 else None
async def _run(cmd: list[str], data: bytes | None) -> bytes | None:
# In a thread, not asyncio's subprocesses: on Windows the client runs a
# selector loop (aiortc's ICE needs one), which cannot start a process.
return await asyncio.to_thread(_run_blocking, cmd, data)
async def read_text() -> str | None:
"""The system clipboard's text, or None when no tool could read it."""
for cmd in _tools(read=True):
out = await _run(cmd, None)
if out is not None:
return out.decode("utf-8", "replace").lstrip("")
return None
async def write_text(text: str) -> bool:
for cmd in _tools(read=False):
if await _run(cmd, _encode(text)) is not None:
return True
return False
|