"""Built-in web admin UI for the middleware. A dependency-free HTTP server (stdlib only) that lets a lab operator configure the cloud connection and up to N analyzer devices from a browser, watch live per-device status, and tail the log — no JSON editing required. Saving applies the new config and hot-restarts the listeners via Application.reload(). Bind defaults to 127.0.0.1 (local only). Set admin.host to 0.0.0.0 in the config to reach it from other machines on the lab network. """ from __future__ import annotations import json import logging import os import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any, Dict from urllib.parse import parse_qs, urlparse from .cloud import CloudClient from .config import KNOWN_DRIVERS, Config log = logging.getLogger("admin") class AdminServer: """Owns the HTTP server thread bound to an Application instance.""" def __init__(self, app: Any) -> None: self.app = app self._httpd: ThreadingHTTPServer | None = None self._thread: threading.Thread | None = None def start(self) -> None: handler = _make_handler(self.app) self._httpd = ThreadingHTTPServer((self.app.cfg.admin_host, self.app.cfg.admin_port), handler) self._thread = threading.Thread(target=self._httpd.serve_forever, name="admin", daemon=True) self._thread.start() def stop(self) -> None: if self._httpd: self._httpd.shutdown() self._httpd.server_close() self._httpd = None def _tail(path: str, n: int) -> str: if not path or not os.path.isfile(path): return "" try: with open(path, "r", encoding="utf-8", errors="replace") as fh: return "".join(fh.readlines()[-n:]) except OSError: return "" def _make_handler(app: Any): class _Handler(BaseHTTPRequestHandler): server_version = "MoonLISMiddleware/1.0" def log_message(self, *args: Any) -> None: # quiet — we have our own logging return # ----- helpers ---------------------------------------------------- def _send(self, code: int, body: bytes, ctype: str) -> None: self.send_response(code) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def _json(self, obj: Any, code: int = 200) -> None: self._send(code, json.dumps(obj, ensure_ascii=False).encode("utf-8"), "application/json; charset=utf-8") def _read_body(self) -> Dict[str, Any]: length = int(self.headers.get("Content-Length") or 0) if not length: return {} raw = self.rfile.read(length) try: return json.loads(raw.decode("utf-8")) except (ValueError, UnicodeDecodeError): return {} # ----- auth ------------------------------------------------------- def _authed(self) -> bool: """Basic-auth gate. Open when no admin password is configured (the UI binds to 127.0.0.1 by default); once a password is set — needed when the UI is exposed to the LAN for multi-station access — every request must present it.""" pw = getattr(app.cfg, "admin_password", "") or "" if not pw: return True hdr = self.headers.get("Authorization", "") if hdr.startswith("Basic "): import base64 try: _, _, passwd = base64.b64decode(hdr[6:]).decode("utf-8").partition(":") if passwd == pw: return True except Exception: # noqa: BLE001 pass self.send_response(401) self.send_header("WWW-Authenticate", 'Basic realm="Moon LIS Middleware"') self.send_header("Content-Length", "0") self.end_headers() return False # ----- routes ----------------------------------------------------- def do_GET(self) -> None: if not self._authed(): return route = urlparse(self.path) path = route.path if path in ("/", "/index.html"): self._send(200, _PAGE.encode("utf-8"), "text/html; charset=utf-8") elif path == "/api/status": self._json(app.status()) elif path == "/api/config": self._json(app.cfg.to_dict()) elif path == "/api/logs": n = int((parse_qs(route.query).get("n") or ["200"])[0]) self._json({"log": _tail(app.cfg.log_file, max(1, min(n, 2000)))}) elif path == "/api/meta": self._json({"drivers": KNOWN_DRIVERS}) else: self._json({"error": "not found"}, 404) def do_POST(self) -> None: if not self._authed(): return path = urlparse(self.path).path body = self._read_body() if path == "/api/config": try: cfg = app.reload(body) self._json({"ok": True, "messages": cfg.validation_messages()}) except Exception as exc: # noqa: BLE001 log.exception("config save failed") self._json({"ok": False, "error": str(exc)}, 500) elif path == "/api/test-cloud": # Test the supplied cloud block WITHOUT saving it. merged = app.cfg.to_dict() merged["cloud"] = body.get("cloud") or merged.get("cloud") result = CloudClient(Config(merged)).test() self._json(result) elif path == "/api/sync": # Pull the device list from LIS → Machines and rebuild listeners. try: summary = app.sync_from_cloud() self._json({"ok": True, "summary": summary}) except Exception as exc: # noqa: BLE001 log.exception("cloud sync failed") self._json({"ok": False, "error": str(exc)}, 500) else: self._json({"error": "not found"}, 404) return _Handler # -------------------------------------------------------------------------- # The single-page dashboard (Arabic-first, RTL). Vanilla JS, no build step. # -------------------------------------------------------------------------- _PAGE = r""" Moon LIS — جسر الأجهزة

🧪 Moon LIS — جسر الأجهزة (Middleware)

هوية المحطة

اتصال Moon ERP (السحابة)

الأجهزة

الأجهزة تُدار من Moon ERP (LIS ◂ الأجهزة) — اضغط «مزامنة» لسحبها هنا ببياناتها تلقائياً.
تشغيلالجهازMachine IDالمنفذالحالةنتائجآخر باركود
جارٍ التحميل…
الحفظ بيكتب الإعدادات ويعيد تشغيل المستمعين فوراً.

السجل (Log) 

…
جهاز
تمرير لبرنامج تاني (Relay) — اختياري
لو في برنامج تاني لازم ياخد نفس الداتا (زي UDI) — نمرّرله البايتات. سيبه فاضي لو مش محتاج.
"""