160 lines
5.5 KiB
Python
160 lines
5.5 KiB
Python
"""HTTP- und WebSocket-Anbindung an den NetClip-Server (ohne Qt, mit Callbacks)."""
|
|
import json
|
|
import ssl
|
|
import struct
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from urllib.parse import quote
|
|
|
|
import requests
|
|
import urllib3
|
|
import websocket
|
|
|
|
from .config import Config
|
|
from .crypto import Keys
|
|
|
|
|
|
class _FramedBody:
|
|
"""Datei-artiger Body: Meta-Laenge + Meta + Blob-Datei, gestreamt mit Fortschritt."""
|
|
|
|
def __init__(self, meta: str, blob: Path, progress=None):
|
|
self.head = struct.pack(">I", len(meta)) + meta.encode("ascii")
|
|
self.file = open(blob, "rb")
|
|
self.len = len(self.head) + blob.stat().st_size
|
|
self.sent = 0
|
|
self.progress = progress
|
|
|
|
def __len__(self):
|
|
return self.len
|
|
|
|
def read(self, n=-1):
|
|
if self.head:
|
|
data, self.head = self.head, b""
|
|
else:
|
|
data = self.file.read(n if n and n > 0 else 1024 * 1024)
|
|
self.sent += len(data)
|
|
if self.progress and data:
|
|
self.progress(self.sent, self.len)
|
|
return data
|
|
|
|
def close(self):
|
|
self.file.close()
|
|
|
|
|
|
class Api:
|
|
def __init__(self, cfg: Config, keys: Keys):
|
|
self.cfg = cfg
|
|
self.keys = keys
|
|
self.base = cfg.server.rstrip("/")
|
|
self.session = requests.Session()
|
|
if cfg.server_key:
|
|
self.session.headers["Authorization"] = f"Bearer {cfg.server_key}"
|
|
if not cfg.verify_tls:
|
|
self.session.verify = False
|
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
|
elif cfg.ca_file:
|
|
self.session.verify = cfg.ca_file
|
|
|
|
@property
|
|
def room_url(self) -> str:
|
|
return f"{self.base}/api/v1/rooms/{self.keys.room_id}"
|
|
|
|
def list_clips(self) -> list[dict]:
|
|
r = self.session.get(f"{self.room_url}/clips", timeout=20)
|
|
r.raise_for_status()
|
|
return r.json()["clips"]
|
|
|
|
def upload(self, meta: dict, blob: Path, progress=None) -> dict:
|
|
body = _FramedBody(self.keys.encrypt_meta(meta), blob, progress)
|
|
try:
|
|
r = self.session.post(f"{self.room_url}/clips", data=body, timeout=(20, 600),
|
|
headers={"X-Device": self.cfg.device_id,
|
|
"Content-Type": "application/octet-stream",
|
|
"Content-Length": str(len(body))})
|
|
finally:
|
|
body.close()
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
def download(self, clip_id: str, dest: Path, progress=None):
|
|
with self.session.get(f"{self.room_url}/clips/{clip_id}/blob", stream=True, timeout=(20, 600)) as r:
|
|
r.raise_for_status()
|
|
total = int(r.headers.get("Content-Length", 0))
|
|
done = 0
|
|
with open(dest, "wb") as f:
|
|
for chunk in r.iter_content(1024 * 1024):
|
|
f.write(chunk)
|
|
done += len(chunk)
|
|
if progress:
|
|
progress(done, total)
|
|
|
|
def delete(self, clip_id: str):
|
|
self.session.delete(f"{self.room_url}/clips/{clip_id}", timeout=20).raise_for_status()
|
|
|
|
def clear(self):
|
|
self.session.delete(f"{self.room_url}/clips", timeout=20).raise_for_status()
|
|
|
|
|
|
class Live(threading.Thread):
|
|
"""WebSocket-Verbindung mit automatischem Wiederverbinden.
|
|
on_message(dict) und on_state(connected: bool, error: str) werden im Thread aufgerufen."""
|
|
|
|
def __init__(self, api: Api, on_message, on_state):
|
|
super().__init__(daemon=True, name="netclip-ws")
|
|
self.api = api
|
|
self.on_message = on_message
|
|
self.on_state = on_state
|
|
self._halt = threading.Event()
|
|
self._ws = None
|
|
|
|
def run(self):
|
|
url = self.api.room_url.replace("https://", "wss://", 1).replace("http://", "ws://", 1)
|
|
url += f"/ws?device={quote(self.api.cfg.device_id)}"
|
|
headers = []
|
|
if self.api.cfg.server_key:
|
|
headers.append(f"Authorization: Bearer {self.api.cfg.server_key}")
|
|
sslopt = {}
|
|
if not self.api.cfg.verify_tls:
|
|
sslopt = {"cert_reqs": ssl.CERT_NONE, "check_hostname": False}
|
|
elif self.api.cfg.ca_file:
|
|
sslopt = {"ca_certs": self.api.cfg.ca_file}
|
|
delay = 2
|
|
while not self._halt.is_set():
|
|
error = ""
|
|
started = time.monotonic()
|
|
self._ws = websocket.WebSocketApp(
|
|
url, header=headers,
|
|
on_open=self._opened,
|
|
on_message=lambda _ws, msg: self._dispatch(msg),
|
|
on_error=lambda _ws, e: None if self._halt.is_set() else self.on_state(False, str(e)),
|
|
)
|
|
try:
|
|
self._ws.run_forever(sslopt=sslopt, ping_interval=30, ping_timeout=10)
|
|
except Exception as e: # noqa: BLE001 - Verbindung soll nie den Thread beenden
|
|
error = str(e)
|
|
if self._halt.is_set():
|
|
return
|
|
self.on_state(False, error)
|
|
if time.monotonic() - started > 60:
|
|
delay = 2
|
|
self._halt.wait(delay)
|
|
delay = min(delay * 2, 60)
|
|
|
|
def _opened(self, ws):
|
|
ws.send(json.dumps({"type": "hello", "label": self.api.keys.encrypt_label(self.api.cfg.device_name)}))
|
|
self.on_state(True, "")
|
|
|
|
def _dispatch(self, raw):
|
|
try:
|
|
self.on_message(json.loads(raw))
|
|
except ValueError:
|
|
pass
|
|
|
|
def stop(self, wait: float = 0):
|
|
self._halt.set()
|
|
if self._ws:
|
|
self._ws.close()
|
|
if wait and self.is_alive():
|
|
self.join(wait)
|