nikola
103 lines
3.1 KiB
Python
103 lines
3.1 KiB
Python
import io
|
|
import os
|
|
import urllib.request
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
|
|
import segno
|
|
|
|
|
|
HOST = "0.0.0.0"
|
|
PORT = 8081
|
|
|
|
HA_BASE = os.getenv("HA_BASE_URL", "http://192.168.0.26:8123").rstrip("/")
|
|
WEBHOOK_ON = os.getenv("HA_WEBHOOK_ON", "guest_light_on_hc_2026_a9f")
|
|
WEBHOOK_OFF = os.getenv("HA_WEBHOOK_OFF", "guest_light_off_hc_2026_b4d")
|
|
|
|
WIFI_SSID = os.getenv("WIFI_SSID", "Zapadna")
|
|
WIFI_PASS = os.getenv("WIFI_PASS", "CHANGE_ME")
|
|
PORTAL_BASE_URL = os.getenv("PORTAL_BASE_URL", "http://192.168.0.26:8090").rstrip("/")
|
|
|
|
|
|
def _svg_bytes(payload: str) -> bytes:
|
|
qr = segno.make(payload)
|
|
buff = io.BytesIO()
|
|
qr.save(buff, kind="svg", scale=6)
|
|
return buff.getvalue()
|
|
|
|
|
|
def _call_webhook(webhook_id: str) -> bool:
|
|
req = urllib.request.Request(
|
|
f"{HA_BASE}/api/webhook/{webhook_id}",
|
|
method="POST",
|
|
data=b"",
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=4) as res:
|
|
return 200 <= res.status < 300
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
if self.path == "/qr/wifi.svg":
|
|
payload = f"WIFI:T:WPA;S:{WIFI_SSID};P:{WIFI_PASS};;"
|
|
data = _svg_bytes(payload)
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "image/svg+xml")
|
|
self.send_header("Cache-Control", "public, max-age=300")
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
return
|
|
|
|
if self.path == "/qr/portal.svg":
|
|
data = _svg_bytes(f"{PORTAL_BASE_URL}/index.html")
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "image/svg+xml")
|
|
self.send_header("Cache-Control", "public, max-age=300")
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
return
|
|
|
|
if self.path == "/qr/bojana.svg":
|
|
data = _svg_bytes(f"{PORTAL_BASE_URL}/bojana.html")
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "image/svg+xml")
|
|
self.send_header("Cache-Control", "public, max-age=300")
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
return
|
|
|
|
if self.path == "/healthz":
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/plain; charset=utf-8")
|
|
self.end_headers()
|
|
self.wfile.write(b"ok")
|
|
return
|
|
|
|
self.send_response(404)
|
|
self.end_headers()
|
|
|
|
def do_POST(self):
|
|
if self.path in ("/api/light/on", "/light/on"):
|
|
ok = _call_webhook(WEBHOOK_ON)
|
|
self.send_response(204 if ok else 502)
|
|
self.end_headers()
|
|
return
|
|
|
|
if self.path in ("/api/light/off", "/light/off"):
|
|
ok = _call_webhook(WEBHOOK_OFF)
|
|
self.send_response(204 if ok else 502)
|
|
self.end_headers()
|
|
return
|
|
|
|
self.send_response(404)
|
|
self.end_headers()
|
|
|
|
def log_message(self, *_):
|
|
return
|
|
|
|
|
|
if __name__ == "__main__":
|
|
HTTPServer((HOST, PORT), Handler).serve_forever()
|