"""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)