clipboard-history/capture.py

#!/usr/bin/env python3
"""Clipboard capture for the alanfortlink.clipboard Omarchy shell plugin.

Invoked by `wl-paste --watch` (payload on stdin is ignored — we probe
ourselves so a single watcher covers every mime) or with no args to snapshot
the current clipboard. Emits exactly one JSON line per capture:

  {"type":"text","text":...,"ts":...,"bytes":...,"app":...}
  {"type":"image","mime":...,"path":...,"w":...,"h":...,"bytes":...,"ts":...,"app":...}
  {"type":"files","paths":[...],"bytes":...,"ts":...,"app":...}

Sensitive clips (x-kde-passwordManagerHint / CLIPBOARD_STATE=sensitive) and
binary payloads are skipped silently.
"""

import hashlib
import json
import os
import shutil
import subprocess
import sys
import tempfile
import time
import urllib.parse

STATE_DIR = os.path.join(
    os.environ.get("XDG_STATE_HOME", os.path.expanduser("~/.local/state")), "omarchy"
)
IMAGE_DIR = os.path.join(STATE_DIR, "clipboard-images")

IMAGE_MIMES = ["image/png", "image/jpeg", "image/webp", "image/gif", "image/bmp", "image/tiff"]
IMAGE_EXT = {"image/png": "png", "image/jpeg": "jpg", "image/webp": "webp",
             "image/gif": "gif", "image/bmp": "bmp", "image/tiff": "tiff"}
TEXT_TYPES = ["text/plain;charset=utf-8", "text/plain", "UTF8_STRING", "STRING", "TEXT", "COMPOUND_TEXT"]


def run(args, timeout=5):
    try:
        r = subprocess.run(args, capture_output=True, timeout=timeout)
        return r.stdout if r.returncode == 0 else None
    except Exception:
        return None


def list_types():
    out = run(["wl-paste", "--list-types"])
    if not out:
        return []
    return [line for line in out.decode("utf-8", "replace").splitlines() if line]


def focused_app():
    out = run(["hyprctl", "activewindow", "-j"], timeout=2)
    if not out:
        return ""
    try:
        data = json.loads(out)
        return str(data.get("class") or "")
    except Exception:
        return ""


def emit(entry):
    entry["ts"] = int(time.time())
    print(json.dumps(entry))
    sys.stdout.flush()


def decode_qr(path):
    """Decode a QR code with zbarimg; returns the payload or None."""
    try:
        r = subprocess.run(
            ["zbarimg", "-q", "--raw", "--", path],
            capture_output=True, timeout=10
        )
        # zbarimg exits 4 when no barcode is found.
        if r.returncode != 0:
            return None
        text = r.stdout.decode("utf-8", "replace").rstrip("\n")
        return text or None
    except Exception:
        return None


def capture_image(types, app):
    mime = next((m for m in IMAGE_MIMES if m in types), None)
    if not mime:
        return
    data = run(["wl-paste", "--type", mime, "--no-newline"])
    if not data:
        return
    os.makedirs(IMAGE_DIR, exist_ok=True)
    digest = hashlib.sha256(data).hexdigest()
    path = os.path.join(IMAGE_DIR, f"{digest}.{IMAGE_EXT[mime]}")
    entry = {"type": "image", "mime": mime, "path": path, "bytes": len(data), "app": app}

    # Dimensions via Pillow when available; harmless without it.
    try:
        from PIL import Image  # noqa: PLC0415
        import io  # noqa: PLC0415
        with Image.open(io.BytesIO(data)) as im:
            entry["w"], entry["h"] = im.size
    except Exception:
        pass

    if not os.path.exists(path):
        fd, tmp = tempfile.mkstemp(dir=IMAGE_DIR)
        try:
            with os.fdopen(fd, "wb") as f:
                f.write(data)
            os.replace(tmp, path)
        except Exception:
            try:
                os.unlink(tmp)
            except OSError:
                pass
            return

    qr = decode_qr(path) if os.environ.get("CLIPBOARD_QR", "1") != "0" else None
    if qr:
        entry["qr"] = qr

    ocr = decode_ocr(path, os.environ.get("CLIPBOARD_OCR_LANG", "eng"))
    if ocr:
        entry["ocr"] = ocr

    emit(entry)
    return True


def decode_text(data):
    """Best-effort text decode; returns None for binary-looking payloads."""
    for enc in ("utf-8", "utf-16"):
        try:
            text = data.decode(enc)
            break
        except (UnicodeDecodeError, UnicodeError):
            continue
    else:
        return None
    # Reject payloads that still look binary after decoding.
    if "\x00" in text:
        return None
    if text:
        nul_control = sum(1 for c in text if ord(c) < 32 and c not in "\n\r\t")
        if nul_control / max(1, len(text)) > 0.05:
            return None
    return text


def decode_ocr(path, lang):
    """Recognize text in an image with tesseract; None when unavailable/binary."""
    if os.environ.get("CLIPBOARD_OCR", "1") == "0":
        return None
    # Tesseract is optional — degrade to no-OCR when missing.
    if shutil.which("tesseract") is None:
        return None
    try:
        r = subprocess.run(
            ["tesseract", path, "stdout", "-l", lang],
            capture_output=True, timeout=30
        )
        if r.returncode != 0:
            return None
        text = r.stdout.decode("utf-8", "replace")
        # Collapse tesseract's whitespace so the stored haystack stays dense.
        text = "\n".join(line.strip() for line in text.splitlines()
                         if line.strip())
        text = text.strip()
        return text[:4000] or None
    except Exception:
        return None


def capture_uri_list(types, app):
    if "text/uri-list" not in types:
        return False
    data = run(["wl-paste", "--type", "text/uri-list", "--no-newline"])
    if not data:
        return False
    text = decode_text(data)
    if text is None:
        return False
    uris = [u.strip() for u in text.splitlines() if u.strip() and not u.startswith("#")]
    if not uris:
        return False
    paths = []
    for uri in uris:
        if uri.startswith("file://"):
            paths.append(urllib.parse.unquote(urllib.parse.urlparse(uri).path))
        else:
            return False  # remote URI → keep it as plain text below
    if not paths:
        return False
    emit({"type": "files", "paths": paths, "bytes": len(data), "app": app})
    return True


def capture_text(types, app):
    mime = next((m for m in TEXT_TYPES if m in types), None)
    if not mime:
        return
    data = run(["wl-paste", "--type", mime, "--no-newline"])
    if not data:
        return
    text = decode_text(data)
    if text is None or not text.strip():
        return
    emit({"type": "text", "text": text, "bytes": len(data), "app": app})


def main():
    # The watcher pipes the clipboard payload to us; wl-clipboard blocks on
    # writes if we close the pipe early, so drain it instead (we still probe
    # types ourselves below).
    try:
        sys.stdin.buffer.read()
    except Exception:
        pass
    types = list_types()
    if not types:
        return
    if "x-kde-passwordManagerHint" in types:
        return
    if os.environ.get("CLIPBOARD_STATE", "") == "sensitive":
        return

    app = focused_app()
    if capture_image(types, app):
        return
    if capture_uri_list(types, app):
        return
    capture_text(types, app)


if __name__ == "__main__":
    main()