#!/usr/bin/env python3 """ 땡구 스마트홈 관리 대시보드 - 서비스 시작/중지/재시작 버튼 - 웹서버 + 음성비서 실시간 로그 - 포트: 9090 """ import http.server import json import os import subprocess import socketserver import threading import time from collections import deque from urllib.parse import urlparse, parse_qs PORT = 9090 SERVICES = { "server": "ddanggu-server", "voice": "ddanggu-voice", } # ── 로그 버퍼 (최근 200줄) ──────────────────── _log_buffers = { "server": deque(maxlen=200), "voice": deque(maxlen=200), } _log_subscribers = { "server": [], "voice": [], } _log_lock = threading.Lock() def _tail_journal(service_key): """journalctl -f 를 백그라운드로 실행, 로그를 버퍼에 저장""" unit = SERVICES[service_key] proc = subprocess.Popen( ["journalctl", "-u", unit, "-f", "-n", "100", "--no-pager", "-o", "short-iso"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True ) for line in proc.stdout: line = line.rstrip() with _log_lock: _log_buffers[service_key].append(line) for q in _log_subscribers[service_key]: q.append(line) def _get_status(unit): try: r = subprocess.run( ["systemctl", "is-active", unit], capture_output=True, text=True, timeout=5 ) return r.stdout.strip() except Exception: return "unknown" def _service_action(unit, action): try: r = subprocess.run( ["sudo", "systemctl", action, unit], capture_output=True, text=True, timeout=30 ) return r.returncode == 0, r.stderr.strip() except Exception as e: return False, str(e) DASHBOARD_HTML = r""" 땡구 관리

땡구 관리 패널

Home Assistant 스마트홈 UI 서버 API
웹서버
음성비서
웹서버 로그 (ddanggu-server)
음성비서 로그 (ddanggu-voice)
""" class Handler(http.server.BaseHTTPRequestHandler): def log_message(self, format, *args): pass # 콘솔 로그 억제 def _json(self, data, code=200): body = json.dumps(data, ensure_ascii=False).encode() self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", len(body)) self.end_headers() self.wfile.write(body) def _html(self, html): body = html.encode() self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", len(body)) self.end_headers() self.wfile.write(body) def do_GET(self): path = urlparse(self.path).path if path == "/": ha_url = os.environ.get("HA_URL", "http://192.168.75.97:8123").rstrip("/") srv_url = os.environ.get("SERVER_URL", "http://192.168.75.97:8080").rstrip("/") html = DASHBOARD_HTML.replace("HA_URL_PLACEHOLDER", ha_url.replace("http://", "")) html = html.replace("SERVER_URL_PLACEHOLDER", srv_url.replace("http://", "")) self._html(html) elif path == "/api/status": self._json({k: _get_status(v) for k, v in SERVICES.items()}) elif path.startswith("/api/logs/"): key = path.split("/")[-1] if key not in _log_buffers: self._json({"error": "unknown service"}, 404) return self.send_response(200) self.send_header("Content-Type", "text/event-stream") self.send_header("Cache-Control", "no-cache") self.send_header("X-Accel-Buffering", "no") self.end_headers() # 기존 버퍼 전송 with _log_lock: for line in _log_buffers[key]: self.wfile.write(f"data: {line}\n\n".encode()) self.wfile.flush() q = deque(maxlen=50) _log_subscribers[key].append(q) try: while True: if q: with _log_lock: while q: line = q.popleft() self.wfile.write(f"data: {line}\n\n".encode()) self.wfile.flush() else: time.sleep(0.3) except (BrokenPipeError, ConnectionResetError, OSError): pass finally: with _log_lock: try: _log_subscribers[key].remove(q) except ValueError: pass else: self._json({"error": "not found"}, 404) def do_POST(self): path = urlparse(self.path).path if path == "/api/control": length = int(self.headers.get("Content-Length", 0)) body = json.loads(self.rfile.read(length)) svc = body.get("service", "") act = body.get("action", "") if act not in ("start", "stop", "restart"): self._json({"ok": False, "message": f"알 수 없는 액션: {act}"}) return if svc == "all": results = [] for key, unit in SERVICES.items(): ok, err = _service_action(unit, act) results.append(f"{key}: {'성공' if ok else err}") self._json({"ok": True, "message": f"전체 {act}: {', '.join(results)}"}) elif svc in SERVICES: ok, err = _service_action(SERVICES[svc], act) label = {"server": "웹서버", "voice": "음성비서"}[svc] msg = f"{label} {act} 성공" if ok else f"{label} {act} 실패: {err}" self._json({"ok": ok, "message": msg}) else: self._json({"ok": False, "message": f"알 수 없는 서비스: {svc}"}) else: self._json({"error": "not found"}, 404) def main(): # settings.env 로드 env_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), "settings.env") if os.path.exists(env_file): with open(env_file, encoding="utf-8") as f: for line in f: line = line.strip() if line and not line.startswith("#") and "=" in line: k, v = line.split("=", 1) os.environ.setdefault(k.strip(), v.strip()) # 로그 스트림 시작 for key in SERVICES: t = threading.Thread(target=_tail_journal, args=(key,), daemon=True) t.start() server = socketserver.ThreadingTCPServer(("", PORT), Handler) server.daemon_threads = True rpi_ip = os.environ.get("RPI_IP", "0.0.0.0") print(f"땡구 관리 패널 → http://{rpi_ip}:{PORT}") try: server.serve_forever() except KeyboardInterrupt: print("\n관리 패널 종료") if __name__ == "__main__": main()