templisandClaude Opus 5
e65f9db3ae
Mirrors the solver's Windows-only voyage-import.ahk on KDE Wayland: - poe-clip-bridge.py: clipboard watcher that keeps only PoE item text, serves it on loopback and can trigger a sweep from the browser - voyage-sweep.py: hovers every cell of the in-game chart grid, Ctrl+C's it and hands over the whole batch in one go - userscript: feeds the batch into the solver page without focusing it Includes notes on the four Wayland pointer pitfalls this ran into. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
192 lines
6.4 KiB
Python
Executable File
192 lines
6.4 KiB
Python
Executable File
#!/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 = """<!doctype html><meta charset=utf-8>
|
||
<title>Voyage sweep</title>
|
||
<style>body{font:14px/1.5 monospace;background:#151515;color:#ddd;margin:2rem}
|
||
h1{font-size:1rem;color:#8f8}pre{white-space:pre-wrap}</style>
|
||
<h1>Voyage sweep</h1><pre id=log>...</pre>
|
||
<script>
|
||
const t = new URLSearchParams(location.search).get('token');
|
||
setInterval(async () => {
|
||
const r = await fetch('/log?token=' + encodeURIComponent(t));
|
||
const d = await r.json();
|
||
document.getElementById('log').textContent =
|
||
d.lines.join('\\n') + (d.running ? '\\n\\n(laeuft...)' : '\\n\\n(fertig)');
|
||
}, 1000);
|
||
</script>"""
|
||
|
||
|
||
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()
|