82 lines
3.5 KiB
Python
82 lines
3.5 KiB
Python
"""Minimal PyQt6 host for dashboard.html (desktop mode). Pattern from Tebra main_window.py.
|
|
|
|
The bot runs as a child process that prints lines, and this window tails them.
|
|
Fill in STEP_MARKERS, FRIENDLY_RULES and BOT_CMD for the project.
|
|
"""
|
|
import json, subprocess, sys, threading
|
|
from pathlib import Path
|
|
from PyQt6.QtCore import QObject, pyqtSlot, pyqtSignal, QUrl, QSettings
|
|
from PyQt6.QtWidgets import QApplication, QMainWindow
|
|
from PyQt6.QtWebEngineWidgets import QWebEngineView
|
|
from PyQt6.QtWebChannel import QWebChannel
|
|
|
|
HTML_PATH = Path(__file__).with_name("dashboard.html")
|
|
BOT_CMD = [sys.executable, "-m", "{{BOT_MODULE}}"] # child process
|
|
STEPS = {{STEPS_PY}} # same list as STEPS in dashboard.html
|
|
STEP_MARKERS = [("[step 1]", 0, 10), ("[all done]", len(STEPS) - 1, 100)] # (substring, step idx, pct)
|
|
FRIENDLY_RULES = [("[login] done", "Logged in.", "success")] # (substring, friendly msg | None, type)
|
|
|
|
|
|
def classify(line):
|
|
s = line.lower()
|
|
return "error" if ("error" in s or "traceback" in s) else "warn" if "warn" in s else "info"
|
|
|
|
|
|
class Bridge(QObject):
|
|
_emit = pyqtSignal(str) # thread-safe hop to the GUI thread
|
|
|
|
def __init__(self, view):
|
|
super().__init__()
|
|
self.view = view
|
|
self._emit.connect(lambda js: self.view.page().runJavaScript(f"window.__pyMsg({js})"))
|
|
|
|
def send_to_js(self, event, data):
|
|
self._emit.emit(json.dumps(json.dumps({"event": event, "data": data})))
|
|
|
|
@pyqtSlot()
|
|
def load_dashboard(self):
|
|
self.send_to_js("dashboard_data", {"batches": []}) # TODO: read from DB on a worker thread
|
|
self.send_to_js("apply_dark", {"on": QSettings("{{ORG}}", "{{APP_NAME}}").value("dark", False, bool)})
|
|
|
|
@pyqtSlot(bool)
|
|
def set_dark(self, on):
|
|
QSettings("{{ORG}}", "{{APP_NAME}}").setValue("dark", on)
|
|
|
|
@pyqtSlot()
|
|
def start_automation(self):
|
|
threading.Thread(target=self._run, daemon=True).start()
|
|
|
|
def _run(self):
|
|
self.send_to_js("status", {"state": "running", "msg": "Running…"})
|
|
p = subprocess.Popen(BOT_CMD, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1)
|
|
for raw in p.stdout:
|
|
self._on_line(raw.rstrip())
|
|
ok = p.wait() == 0
|
|
self.send_to_js("status", {"state": "done" if ok else "error", "msg": "Completed" if ok else "Failed"})
|
|
|
|
def _on_line(self, line):
|
|
# ponytail: raw line only printed here; persist to file/DB if audit is needed
|
|
low, kind = line.lower(), classify(line)
|
|
for marker, idx, pct in STEP_MARKERS:
|
|
if marker in low:
|
|
self.send_to_js("step", {"label": STEPS[idx], "pct": pct})
|
|
break
|
|
for marker, msg, t in FRIENDLY_RULES:
|
|
if marker in low:
|
|
if msg:
|
|
self.send_to_js("log", {"msg": msg, "type": t})
|
|
return
|
|
if kind == "error": # never hide a real failure
|
|
self.send_to_js("log", {"msg": line, "type": "error"})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app = QApplication(sys.argv)
|
|
win, view = QMainWindow(), QWebEngineView()
|
|
bridge, channel = Bridge(view), QWebChannel()
|
|
channel.registerObject("pyBridge", bridge)
|
|
view.page().setWebChannel(channel)
|
|
view.setUrl(QUrl.fromLocalFile(str(HTML_PATH)))
|
|
win.setCentralWidget(view); win.resize(1280, 820); win.setWindowTitle("{{APP_NAME}}"); win.show()
|
|
sys.exit(app.exec())
|