Merge pull request #19 from benfirad/codex/macos-clipboard-native

Preserve every native macOS clipboard format

paste.py

@@ -11,10 +11,12 @@ every function here.
 import collections
 import ctypes
 import functools
+import json
 import os
 import shutil
 import subprocess
 import sys
+import tempfile
 import time
 
 from i18n import t
@@ -49,6 +51,64 @@ MAC_ALIASES = {"cmd": "command", "meta": "command", "super": "command",
 HID_EVENT_TAP = 0   # kCGHIDEventTap: the event goes in where the keyboard does
 
 
+# pbpaste only reads text, EPS and RTF.  In particular, an image on a Mac's
+# clipboard comes back as an empty byte string and pbcopy then replaces it with
+# empty plain text.  Keep every NSPasteboard representation in short-lived
+# files instead.  The manifest stays small even when the clipboard holds a
+# large TIFF, and no additional Python package is needed.
+_MAC_SNAPSHOT = collections.namedtuple("MacClipboardSnapshot", "directory manifest")
+
+_MAC_SNAPSHOT_SCRIPT = r'''
+ObjC.import("AppKit");
+const root = ObjC.unwrap(
+  $.NSProcessInfo.processInfo.environment.objectForKey("DIKTE_PASTEBOARD_DIR")
+);
+const pasteboard = $.NSPasteboard.generalPasteboard;
+const items = pasteboard.pasteboardItems;
+const result = [];
+for (let i = 0; i < items.count; i++) {
+  const item = items.objectAtIndex(i);
+  const representations = [];
+  const types = item.types;
+  for (let j = 0; j < types.count; j++) {
+    const type = ObjC.unwrap(types.objectAtIndex(j));
+    const data = item.dataForType(type);
+    if (!data) continue;
+    const file = `${i}-${j}.bin`;
+    if (data.writeToFileAtomically(`${root}/${file}`, true)) {
+      representations.push({type, file});
+    }
+  }
+  result.push(representations);
+}
+JSON.stringify(result);
+'''
+
+_MAC_RESTORE_SCRIPT = r'''
+ObjC.import("AppKit");
+const root = ObjC.unwrap(
+  $.NSProcessInfo.processInfo.environment.objectForKey("DIKTE_PASTEBOARD_DIR")
+);
+const input = $.NSFileHandle.fileHandleWithStandardInput.readDataToEndOfFile;
+const source = $.NSString.alloc.initWithDataEncoding(input, $.NSUTF8StringEncoding);
+const rows = JSON.parse(ObjC.unwrap(source));
+const items = [];
+for (const representations of rows) {
+  const item = $.NSPasteboardItem.alloc.init;
+  for (const representation of representations) {
+    const data = $.NSData.dataWithContentsOfFile(
+      `${root}/${representation.file}`
+    );
+    if (data) item.setDataForType(data, representation.type);
+  }
+  items.push(item);
+}
+const pasteboard = $.NSPasteboard.generalPasteboard;
+pasteboard.clearContents;
+pasteboard.writeObjects($(items));
+'''
+
+
 class PasteError(Exception):
     pass
 
@@ -280,8 +340,45 @@ def desktop():
 
 # --- the clipboard ---------------------------------------------------------
 
+def _macos_snapshot():
+    """Copy every native pasteboard type to a temporary, file-backed snapshot."""
+    directory = tempfile.mkdtemp(prefix="dikte-clipboard-")
+    environment = dict(os.environ, DIKTE_PASTEBOARD_DIR=directory)
+    try:
+        result = subprocess.run(
+            ["osascript", "-l", "JavaScript", "-e", _MAC_SNAPSHOT_SCRIPT],
+            capture_output=True, text=True, timeout=15, env=environment,
+        )
+        manifest = result.stdout.strip()
+        rows = json.loads(manifest) if result.returncode == 0 else None
+        if not isinstance(rows, list):
+            raise ValueError("the pasteboard helper returned no manifest")
+        return _MAC_SNAPSHOT(directory, manifest)
+    except (json.JSONDecodeError, OSError, subprocess.SubprocessError, ValueError):
+        shutil.rmtree(directory, ignore_errors=True)
+        return None
+
+
+def _macos_restore(snapshot):
+    """Put a native snapshot back, then discard its short-lived files."""
+    environment = dict(os.environ, DIKTE_PASTEBOARD_DIR=snapshot.directory)
+    try:
+        subprocess.run(
+            ["osascript", "-l", "JavaScript", "-e", _MAC_RESTORE_SCRIPT],
+            input=snapshot.manifest, stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL, text=True, timeout=15, env=environment,
+        )
+    except (OSError, subprocess.SubprocessError):
+        pass
+    finally:
+        shutil.rmtree(snapshot.directory, ignore_errors=True)
+
 def read_clipboard():
     here = desktop()
+    if here is MACOS and shutil.which("osascript"):
+        snapshot = _macos_snapshot()
+        if snapshot is not None:
+            return snapshot
     if not shutil.which(here.read_command[0]):
         return None
     try:
@@ -321,6 +418,9 @@ def copy(text):
 
 
 def copy_bytes(data):
+    if isinstance(data, _MAC_SNAPSHOT):
+        _macos_restore(data)
+        return
     if data is None or not shutil.which(desktop().clipboard):
         return
     try:

tests/test_paste.py

@@ -14,8 +14,10 @@ cannot quietly break the platform nobody is sitting at.
 """
 
 import os
+import pathlib
 import subprocess
 import sys
+import tempfile
 import unittest
 from typing import ClassVar
 from unittest import mock
@@ -402,5 +404,30 @@ class MacOS(ClipboardContract, DikteTest):
         self.assertFalse(paste.paste_ready())
 
 
+class MacClipboardSnapshot(DikteTest):
+    def test_every_native_type_is_restored_and_the_files_are_removed(self):
+        directory = tempfile.mkdtemp(prefix="dikte-test-clipboard-")
+        manifest = '[[{"type":"public.tiff","file":"0-0.bin"}]]'
+        pathlib.Path(directory, "0-0.bin").write_bytes(b"a TIFF")
+        snapshot = paste._MAC_SNAPSHOT(directory, manifest)
+
+        with mock.patch.object(subprocess, "run",
+                               return_value=FakeCompleted()) as run:
+            paste.copy_bytes(snapshot)
+
+        self.assertEqual(run.call_args.kwargs["input"], manifest)
+        self.assertEqual(run.call_args.kwargs["env"]["DIKTE_PASTEBOARD_DIR"],
+                         directory)
+        self.assertFalse(os.path.exists(directory))
+
+    def test_a_failed_snapshot_leaves_no_temporary_directory(self):
+        directory = tempfile.mkdtemp(prefix="dikte-test-clipboard-")
+        with mock.patch.object(paste.tempfile, "mkdtemp", return_value=directory), \
+                mock.patch.object(subprocess, "run",
+                                  return_value=FakeCompleted(stdout=b"not json")):
+            self.assertIsNone(paste._macos_snapshot())
+        self.assertFalse(os.path.exists(directory))
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_worker.py

@@ -32,7 +32,7 @@ class Chain(DikteTest):
                   transcript="uh, book it for Thursday",
                   cleaned="Book it for Thursday.",
                   cleanup_error=None, answer=("Booked.", ""), rms=None,
-                  clipboard=b"what was there before"):
+                  clipboard=b"what was there before", paste_error=None):
         pipeline = worker.Pipeline(self.conf)
         done, failures, stages, cancels = [], [], [], []
         pipeline.finished.connect(lambda *args: done.append(args))
@@ -52,10 +52,13 @@ class Chain(DikteTest):
                 mock.patch.object(paste, "copy") as copy, \
                 mock.patch.object(paste, "copy_bytes") as copy_bytes, \
                 mock.patch.object(paste, "press") as press, \
-                mock.patch.object(paste, "read_clipboard", return_value=clipboard), \
+                mock.patch.object(paste, "read_clipboard",
+                                  return_value=clipboard) as read_clipboard, \
                 mock.patch.object(worker.time, "sleep", lambda seconds: None):
+            press.side_effect = paste_error
             calls = {"transcribe": tr, "cleanup": cleanup, "ask": ask_call,
-                     "copy": copy, "copy_bytes": copy_bytes, "press": press}
+                     "copy": copy, "copy_bytes": copy_bytes, "press": press,
+                     "read_clipboard": read_clipboard}
             pipeline._work(self.wav, duration,
                            self.rms if rms is None else rms, ask, paste_override)
         return {"done": done, "failures": failures, "stages": stages,
@@ -83,9 +86,11 @@ class Chain(DikteTest):
 
     def test_auto_paste_switched_off_only_copies(self):
         self.conf["auto_paste"] = False
+        self.conf["restore_clipboard"] = True
         run = self.run_chain()
         run["copy"].assert_called_once()
         run["press"].assert_not_called()
+        run["read_clipboard"].assert_not_called()
 
     def test_a_run_asked_for_from_a_terminal_pastes_nowhere(self):
         """The text comes back down the socket; the focused window is nobody's."""
@@ -103,6 +108,12 @@ class Chain(DikteTest):
         run = self.run_chain()
         run["copy_bytes"].assert_not_called()
 
+    def test_the_clipboard_is_put_back_when_the_keypress_fails(self):
+        self.conf["restore_clipboard"] = True
+        run = self.run_chain(paste_error=paste.PasteError("not trusted"))
+        self.assertIn("not trusted", run["failures"][0])
+        run["copy_bytes"].assert_called_once_with(b"what was there before")
+
     def test_the_transcription_is_told_the_language_and_the_glossary(self):
         self.conf["language"] = "tr"
         self.conf["transcribe_prompt"] = "Paraşüt"

worker.py

@@ -138,13 +138,18 @@ class Pipeline(QObject):
                 wants_paste = paste_override
 
             with _paste_lock:
-                previous = paste.read_clipboard() if conf["restore_clipboard"] else None
-                paste.copy(text)
-
-                if wants_paste:
-                    self.stage.emit(t("Pasting…"))
-                    paste.press(conf["paste_shortcut"])
+                previous = (paste.read_clipboard()
+                            if conf["restore_clipboard"] and wants_paste else None)
+                try:
+                    paste.copy(text)
+                    if wants_paste:
+                        self.stage.emit(t("Pasting…"))
+                        paste.press(conf["paste_shortcut"])
+                finally:
                     if previous is not None:
+                        # Let the focused application consume the temporary
+                        # transcription before putting every old clipboard type
+                        # back.  This also runs when key injection fails.
                         time.sleep(0.35)
                         paste.copy_bytes(previous)