#!/usr/bin/env python3
"""Clipboard bridge for the Allflame Voyage Solver.
Watches the Wayland clipboard, keeps ONLY Path of Exile item text in memory
and serves it on http://127.0.0.1:8477/clip for a Violentmonkey userscript.
Also triggers the grid sweep, so a Firefox bookmark can start it.
Security: anything that does not look like PoE item text (passwords from
KeePassXC, tokens, ...) is discarded immediately and never stored or served.
Nothing is written to disk. The socket is bound to loopback only.
Endpoints that act (sweep/pause/resume) require the token from
~/.config/poe-clip-bridge/token - otherwise any web page could start a sweep
that takes over mouse and keyboard.
"""
import json
import os
import pathlib
import re
import secrets
import subprocess
import threading
import urllib.parse
from collections import deque
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
HOST, PORT = "127.0.0.1", 8477
ORIGIN = "https://one-more-map.github.io"
TOKEN_FILE = pathlib.Path.home() / ".config" / "poe-clip-bridge" / "token"
SWEEP = str(pathlib.Path.home() / ".local" / "bin" / "voyage-sweep.py")
# Only clipboard content matching this ever leaves the watcher.
ITEM_RE = re.compile(
r"^[ \t]*(?:Item Class|Rarity)\s*[:：]|===\s*VOYAGE BORDER",
re.IGNORECASE | re.MULTILINE,
)
state = {"seq": 0, "text": ""}
paused = False
sweep_proc = None
sweep_log = deque(maxlen=40)
lock = threading.Lock()
def load_token() -> str:
if TOKEN_FILE.exists():
return TOKEN_FILE.read_text().strip()
TOKEN_FILE.parent.mkdir(parents=True, exist_ok=True)
token = secrets.token_urlsafe(24)
TOKEN_FILE.write_text(token + "\n")
TOKEN_FILE.chmod(0o600)
return token
TOKEN = load_token()
def publish(text: str) -> None:
"""Store item text; called by the watcher and once again on resume."""
if not ITEM_RE.search(text):
return
with lock:
if text != state["text"]:
state["seq"] += 1
state["text"] = text
print(f"[bridge] item #{state['seq']} ({len(text)} chars)", flush=True)
def watch() -> None:
"""Stream clipboard changes as NUL-delimited records from wl-paste."""
proc = subprocess.Popen(
["wl-paste", "--watch", "sh", "-c", "cat; printf '\\0'"],
stdout=subprocess.PIPE,
)
buf = b""
while True:
chunk = proc.stdout.read1(4096) # read1: do not block for a full buffer
if not chunk:
break
buf += chunk
while b"\0" in buf:
raw, buf = buf.split(b"\0", 1)
if paused: # during a sweep every single copy would import separately
continue
publish(raw.decode("utf-8", "replace").strip())
def read_clipboard() -> str:
r = subprocess.run(["wl-paste", "--no-newline", "--type", "text/plain"],
capture_output=True, text=True)
return r.stdout.strip() if r.returncode == 0 else ""
def run_sweep() -> None:
"""Run the grid sweep, capturing its output for the status page."""
global sweep_proc, paused
paused = True
sweep_log.clear()
sweep_log.append("sweep started")
try:
sweep_proc = subprocess.Popen(
["python3", "-u", SWEEP, "run", "--delay", "4"],
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
)
for line in sweep_proc.stdout:
for part in line.replace("\r", "\n").splitlines():
if part.strip():
sweep_log.append(part.strip())
sweep_proc.wait()
sweep_log.append(f"sweep finished (exit {sweep_proc.returncode})")
except Exception as exc: # noqa: BLE001 - surfaced on the status page
sweep_log.append(f"sweep failed: {exc}")
finally:
sweep_proc = None
paused = False
publish(read_clipboard()) # the batch the sweep just put there
STATUS_PAGE = """
Voyage sweep
Voyage sweep
...
"""
class Handler(BaseHTTPRequestHandler):
def _send(self, body, ctype="application/json", code=200):
data = body.encode() if isinstance(body, str) else body
self.send_response(code)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(data)))
self.send_header("Access-Control-Allow-Origin", ORIGIN)
self.send_header("Cache-Control", "no-store")
self.end_headers()
self.wfile.write(data)
def _authed(self, query):
return secrets.compare_digest(query.get("token", [""])[0], TOKEN)
def do_GET(self):
parsed = urllib.parse.urlparse(self.path)
query = urllib.parse.parse_qs(parsed.query)
path = parsed.path
if path == "/clip":
with lock:
self._send(json.dumps(state))
return
if path in ("/sweep", "/log", "/pause", "/resume"):
if not self._authed(query):
self._send('{"error":"bad token"}', code=403)
return
global paused
if path == "/sweep":
if sweep_proc is None:
threading.Thread(target=run_sweep, daemon=True).start()
self._send(STATUS_PAGE, ctype="text/html; charset=utf-8")
elif path == "/log":
self._send(json.dumps({"running": sweep_proc is not None,
"lines": list(sweep_log)}))
elif path == "/pause":
paused = True
self._send('{"paused":true}')
elif path == "/resume":
paused = False
publish(read_clipboard())
self._send('{"paused":false}')
else:
self.send_error(404)
def log_message(self, *args):
pass
if __name__ == "__main__":
threading.Thread(target=watch, daemon=True).start()
print(f"[bridge] listening on http://{HOST}:{PORT}/clip", flush=True)
print(f"[bridge] sweep bookmark: http://{HOST}:{PORT}/sweep?token={TOKEN}",
flush=True)
ThreadingHTTPServer((HOST, PORT), Handler).serve_forever()