sanspie/homelab-radio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
homelab-radio/sdr/web/handler.py
T

206 lines
8.8 KiB
Python
Raw Permalink Blame History

"""HTTP request handler: small routing tables (GET by exact path / prefix, POST by path)."""
import json
import re
from http.server import BaseHTTPRequestHandler
from urllib.parse import parse_qs, urlparse
from radio.config import HOME_PROXY, LAN_PREFIX, MAX_LIST, PORT, SCANNER_UNIT
from radio.recordings import counts, find_item, scan
from web import audio, events
from web.pages import FAVICON, SHARE_PAGE, page
from web.shares import share_conv, share_token
from web.sources import adsb_latest, promote, sources
from web.speaker import speaker_set, speaker_state
from web.status import status, transcribe_status, unit_state
JSON_UTF8 = "application/json; charset=utf-8"
def qint(qs, k, d, lo, hi):
try:
return max(lo, min(hi, int(qs.get(k, [d])[0])))
except ValueError:
return d
class H(BaseHTTPRequestHandler):
server_version = "sdr-web/2"
protocol_version = "HTTP/1.1"
def log_message(self, fmt, *a):
pass
# ---- helpers ------------------------------------------------------------------------------
def send(self, code, body, ctype="text/html; charset=utf-8", extra=None):
b = body.encode("utf-8") if isinstance(body, str) else body
self.send_response(code)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(b)))
self.send_header("Cache-Control", "no-store")
for k, v in (extra or {}).items():
self.send_header(k, v)
self.end_headers()
if self.command != "HEAD":
self.wfile.write(b)
def from_lan(self):
"""Client is on the home LAN: its own address, or (via the home Caddy) the X-Home-Client it passes on."""
ip = self.client_address[0]
if ip == HOME_PROXY:
ip = (self.headers.get("X-Home-Client") or "").strip()
return ip.startswith(LAN_PREFIX)
def js(self, data, cors=True):
return self.send(200, json.dumps(data, ensure_ascii=False), JSON_UTF8,
{"Access-Control-Allow-Origin": "*"} if cors else None)
# ---- GET routes: handler(self, qs) --------------------------------------------------------
def r_index(self, qs):
return self.send(200, page())
def r_favicon(self, qs):
return self.send(200, FAVICON, "image/svg+xml", {"Cache-Control": "public, max-age=604800"})
def r_list(self, qs):
voice = qs.get("voice", ["1"])[0] != "0"
gs = [g for g in qs.get("groups", [""])[0].split(",") if g] or None
before = qs.get("before", [""])[0] or None
chan = re.sub(r"\D", "", qs.get("chan", [""])[0]) or None
items, pending = scan(qint(qs, "limit", 400, 1, MAX_LIST), voice, None, "", True, groups=gs, before=before, chan=chan)
return self.js({"items": items, "pending": pending, "status": status(), "counts": counts(voice),
"more": len(items) >= qint(qs, "limit", 400, 1, MAX_LIST)}, cors=False)
def r_recent(self, qs):
g = qs.get("group", [None])[0]
noise = qs.get("noise", ["0"])[0] == "1"
host = self.headers.get("Host") or f"192.168.88.35:{PORT}"
items, _ = scan(qint(qs, "limit", 20, 1, 500), not noise, g,
"http://" + re.sub(r"[^A-Za-z0-9.:\[\]-]", "", host))
return self.js(items)
def r_status(self, qs):
return self.js(status())
def r_transcribe(self, qs):
"""For the home dashboard's health check: 200 while Whisper on the desktop answers, else 503."""
t = transcribe_status() or {"stale": True}
ok = t.get("endpoint_up") and not t.get("stale")
return self.send(200 if ok else 503, json.dumps(t, ensure_ascii=False), JSON_UTF8)
def r_adsb(self, qs):
return self.js(adsb_latest(), cors=False)
def r_sources(self, qs):
return self.js(sources(), cors=False)
def r_item(self, qs):
it = find_item(qs.get("id", [""])[0])
if it is None:
return self.send(404, json.dumps({"error": "gone"}), JSON_UTF8)
return self.js(it, cors=False)
def r_events(self, qs):
return events.stream(self)
def r_speaker(self, qs):
return self.js(dict(speaker_state(), local=self.from_lan()), cors=False)
def r_mode(self, qs):
# legacy alias, kept for old clients
s = unit_state()
return self.js({"mode": "scanner" if s in ("active", "activating", "reloading") else "off",
"units": {SCANNER_UNIT: s}, "switching": False})
def r_share(self, qs, path):
return self.share_get(path.split("/")[2:], qs.get("dl", ["0"])[0] == "1")
def r_audio(self, qs, path):
return audio.serve(self, path[len("/audio/"):], qs.get("dl", ["0"])[0] == "1")
GET_EXACT = {
"/": r_index, "/favicon.svg": r_favicon, "/favicon.ico": r_favicon,
"/api/list": r_list, "/api/recent": r_recent, "/api/status": r_status, "/api/transcribe": r_transcribe, "/api/adsb": r_adsb,
"/api/sources": r_sources, "/api/item": r_item, "/api/events": r_events,
"/api/speaker": r_speaker, "/api/mode": r_mode,
}
GET_PREFIX = (("/s/", r_share), ("/audio/", r_audio))
def do_HEAD(self):
self.do_GET()
def do_GET(self):
u = urlparse(self.path)
qs = parse_qs(u.query)
try:
fn = self.GET_EXACT.get(u.path)
if fn:
return fn(self, qs)
for prefix, fn in self.GET_PREFIX:
if u.path.startswith(prefix):
return fn(self, qs, u.path)
return self.send(404, "not found", "text/plain")
except (BrokenPipeError, ConnectionResetError):
pass
def share_get(self, parts, dl):
conv = share_conv(parts[0] if parts else "")
if conv is None:
if len(parts) == 1:
return self.send(200, SHARE_PAGE, extra={"X-Robots-Tag": "noindex"})
return self.send(404, json.dumps({"error": "gone"}), JSON_UTF8)
items, me = conv
if len(parts) == 1:
return self.send(200, SHARE_PAGE, extra={"X-Robots-Tag": "noindex"})
if parts[1:] == ["data"]:
keep = ("channel", "group", "freq_mhz", "ts", "duration_s", "peaks")
return self.js({"me": me, "items": [{k: x.get(k) for k in keep} for x in items]}, cors=False)
if len(parts) == 3 and parts[1] == "a" and parts[2].isdigit() and int(parts[2]) < len(items):
return audio.serve(self, items[int(parts[2])]["file"], dl)
return self.send(404, "not found", "text/plain")
# ---- POST routes: handler(self, body) -----------------------------------------------------
def p_share(self, body):
try:
rid = str(json.loads(body.decode("utf-8") or "{}").get("id") or "")
except (ValueError, AttributeError):
rid = ""
it = find_item(rid)
if not it:
return self.send(404, json.dumps({"ok": False, "msg": "запись не найдена"}, ensure_ascii=False), JSON_UTF8)
return self.js({"ok": True, "url": f"/s/{share_token(it['id'])}"}, cors=False)
def p_speaker(self, body):
if not self.from_lan():
return self.send(403, json.dumps({"ok": False, "msg": "только из домашней сети"}, ensure_ascii=False),
JSON_UTF8)
try:
b = json.loads(body.decode("utf-8") or "{}")
st = speaker_set(b.get("on"), b.get("volume"))
except (ValueError, TypeError, AttributeError, OSError) as e:
return self.send(400, json.dumps({"ok": False, "msg": str(e)}), "application/json")
return self.js(dict(st, local=True), cors=False)
def p_promote(self, body):
try:
freq = int(json.loads(body.decode("utf-8") or "{}").get("f"))
except (ValueError, TypeError, AttributeError):
return self.send(400, json.dumps({"ok": False, "msg": "bad request"}), "application/json")
code, msg = promote(freq)
return self.send(code, json.dumps({"ok": code == 200, "msg": msg}, ensure_ascii=False), JSON_UTF8)
POST_ROUTES = {"/api/promote": p_promote, "/api/speaker": p_speaker, "/api/share": p_share}
def do_POST(self):
# mode switching was removed on purpose: the radio site is recordings-only.
# Writes: queue a discovered frequency, speaker switch (LAN only), create a share link.
u = urlparse(self.path)
n = min(int(self.headers.get("Content-Length") or 0), 4096)
body = self.rfile.read(n) if n else b""
fn = self.POST_ROUTES.get(u.path)
if fn is None:
return self.send(405, "method not allowed", "text/plain", {"Allow": "GET, HEAD"})
origin = self.headers.get("Origin")
if origin and urlparse(origin).netloc != self.headers.get("Host", ""):
return self.send(403, "forbidden", "text/plain")
return fn(self, body)
Reference in New Issue View Git Blame Copy Permalink