340 lines
14 KiB
Python
340 lines
14 KiB
Python
"""
|
|
Quant-UX REST API client — core library shared by the MCP server and CLI.
|
|
|
|
Talks directly to the Quant-UX backend through its REST API
|
|
(frontend proxies /rest/* to the Java backend).
|
|
|
|
Key facts learned from reading qux-java source:
|
|
* Auth: POST /rest/user (register), POST /rest/login (returns {"token": <JWT>})
|
|
subsequent calls: Authorization: Bearer <JWT>
|
|
* Apps: GET /rest/apps, POST /rest/apps, GET /rest/apps/:id.json, DELETE ...
|
|
* Changes: POST /rest/apps/:id/update body = JSON array of deltas:
|
|
{"type": "add"|"update"|"delete",
|
|
"name": <field or id>,
|
|
"parent": "screens"|"widgets"|"lines"|"groups"|"templates"|null,
|
|
"object": <value>}
|
|
-> translated server-side into mongo $set/$unset
|
|
* Pitfall: the backend rejects the payload with HTTP 405 if it does not
|
|
start with "[" AND end with "]" (no trailing newline allowed).
|
|
requests' json= parameter serializes compact JSON without a
|
|
trailing newline, so it is safe.
|
|
* Model format:
|
|
screens: {"<id>": {id,name,x,y,w,h,z,min,props:{start},style,has,children:[...]}}
|
|
widgets: {"<id>": {id,name,type,x,y,w,h,z,props,has,actions,style}}
|
|
lines: {"<id>": {id,from,to,event,points}}
|
|
model-level: name, description, type, screenSize{w,h}, startScreen, lastUUID, grid
|
|
* Widget MUST carry a non-null "style" (frontend ModelFixer deletes widgets
|
|
without style).
|
|
"""
|
|
|
|
import base64
|
|
import binascii
|
|
import json
|
|
import re
|
|
import time
|
|
|
|
import requests
|
|
|
|
REST_BASE = "/rest" # kept for reference; paths below include it explicitly
|
|
|
|
|
|
class QuantUXError(Exception):
|
|
"""Raised for any API-level failure."""
|
|
|
|
|
|
def _default_style(widget_type):
|
|
"""Sensible defaults per widget type (mirrors what the frontend uses)."""
|
|
font = "Helvetica Neue,Helvetica,Arial,sans-serif"
|
|
border_zero = {
|
|
"borderTopWidth": 0, "borderBottomWidth": 0,
|
|
"borderRightWidth": 0, "borderLeftWidth": 0,
|
|
"borderTopColor": "#000000", "borderBottomColor": "#000000",
|
|
"borderRightColor": "#000000", "borderLeftColor": "#000000",
|
|
}
|
|
radius_zero = {
|
|
"borderTopRightRadius": 0, "borderTopLeftRadius": 0,
|
|
"borderBottomRightRadius": 0, "borderBottomLeftRadius": 0,
|
|
}
|
|
if widget_type == "Box":
|
|
return {**border_zero, "background": "#E5E7EB"}
|
|
if widget_type == "Label":
|
|
return {
|
|
"fontSize": 16, "fontFamily": font, "textAlign": "left",
|
|
"letterSpacing": 0, "lineHeight": 1.4, "color": "#111827",
|
|
"textShadow": None,
|
|
}
|
|
if widget_type in ("TextBox", "Password", "TextArea"):
|
|
return {
|
|
**border_zero, **radius_zero,
|
|
"borderTopWidth": 1, "borderBottomWidth": 1,
|
|
"borderRightWidth": 1, "borderLeftWidth": 1,
|
|
"borderTopColor": "#D1D5DB", "borderBottomColor": "#D1D5DB",
|
|
"borderRightColor": "#D1D5DB", "borderLeftColor": "#D1D5DB",
|
|
"background": "#FFFFFF", "fontSize": 14, "color": "#111827",
|
|
"paddingLeft": 12, "paddingRight": 12,
|
|
"paddingTop": 0, "paddingBottom": 0, "textShadow": None,
|
|
}
|
|
if widget_type == "Button":
|
|
return {
|
|
"fontSize": 14, "fontFamily": font, "textAlign": "center",
|
|
"letterSpacing": 0, "lineHeight": 1.4, "color": "#FFFFFF",
|
|
**radius_zero, **border_zero, "background": "#111827",
|
|
"paddingTop": 0, "paddingBottom": 0,
|
|
"paddingLeft": 0, "paddingRight": 0, "textShadow": None,
|
|
}
|
|
if widget_type == "HotSpot":
|
|
return {}
|
|
# generic fallback
|
|
return {**border_zero, **radius_zero, "background": "#FFFFFF"}
|
|
|
|
|
|
def _default_has(widget_type):
|
|
if widget_type == "Label":
|
|
return {"label": True, "padding": True, "advancedText": True}
|
|
if widget_type in ("TextBox", "Password", "TextArea"):
|
|
return {"label": True, "border": True, "padding": True, "backgroundColor": True}
|
|
if widget_type == "Button":
|
|
return {"backgroundColor": True, "border": True, "label": True,
|
|
"padding": True, "onclick": True}
|
|
if widget_type == "HotSpot":
|
|
return {"onclick": True}
|
|
return {"backgroundColor": True, "border": True}
|
|
|
|
|
|
class QuantUXClient:
|
|
"""Thin, battle-tested wrapper around the Quant-UX REST API."""
|
|
|
|
def __init__(self, base_url, token=None, timeout=30):
|
|
self.base_url = base_url.rstrip("/")
|
|
self.timeout = timeout
|
|
self.session = requests.Session()
|
|
if token:
|
|
self.token = token
|
|
self.session.headers["Authorization"] = f"Bearer {token}"
|
|
else:
|
|
self.token = None
|
|
self._auth_refresh_handler = None
|
|
|
|
def set_auth_refresh_handler(self, handler):
|
|
"""Register a callback used once after an explicit 401/403 response."""
|
|
self._auth_refresh_handler = handler
|
|
|
|
def token_expires_at(self):
|
|
"""Return the JWT exp timestamp without logging or verifying the token."""
|
|
if not self.token:
|
|
return None
|
|
try:
|
|
payload = self.token.split(".")[1]
|
|
payload += "=" * (-len(payload) % 4)
|
|
decoded = json.loads(base64.urlsafe_b64decode(payload.encode("ascii")))
|
|
expires_at = decoded.get("exp")
|
|
return int(expires_at) if expires_at is not None else None
|
|
except (IndexError, ValueError, TypeError, binascii.Error):
|
|
return None
|
|
|
|
def token_is_valid(self, leeway=60, now=None):
|
|
expires_at = self.token_expires_at()
|
|
current = time.time() if now is None else now
|
|
return bool(expires_at and expires_at > current + max(0, leeway))
|
|
|
|
# ------------------------------------------------------------------ auth
|
|
def login(self, email, password):
|
|
"""Login and cache the JWT for all subsequent calls."""
|
|
r = self._req(
|
|
"POST",
|
|
"/rest/login",
|
|
json={"email": email, "password": password},
|
|
_allow_auth_retry=False,
|
|
)
|
|
token = r.get("token")
|
|
if not token:
|
|
raise QuantUXError("Login succeeded but no token in response")
|
|
self.token = token
|
|
self.session.headers["Authorization"] = f"Bearer {token}"
|
|
return r
|
|
|
|
def register(self, name, lastname, email, password):
|
|
return self._req("POST", "/rest/user", json={
|
|
"name": name, "lastname": lastname,
|
|
"email": email, "password": password, "tos": True,
|
|
})
|
|
|
|
# ------------------------------------------------------------------ apps
|
|
def list_apps(self):
|
|
return self._req("GET", "/rest/apps")
|
|
|
|
def get_app(self, app_id):
|
|
return self._req("GET", f"/rest/apps/{app_id}.json")
|
|
|
|
def create_app(self, name, description="", width=375, height=667,
|
|
app_type="prototype", is_public=False):
|
|
r = self._req("POST", "/rest/apps", json={
|
|
"name": name,
|
|
"description": description,
|
|
"type": app_type,
|
|
"screenSize": {"w": width, "h": height},
|
|
"isPublic": is_public,
|
|
})
|
|
return r["_id"]
|
|
|
|
def delete_app(self, app_id):
|
|
return self._req("DELETE", f"/rest/apps/{app_id}.json")
|
|
|
|
# ---------------------------------------------------------------- changes
|
|
def apply_changes(self, app_id, changes):
|
|
"""POST a delta array to /rest/apps/:id/update (the only write path)."""
|
|
if not isinstance(changes, list):
|
|
raise QuantUXError("changes must be a JSON array")
|
|
return self._req("POST", f"/rest/apps/{app_id}/update", json=changes)
|
|
|
|
def _next_id(self, model):
|
|
"""Next numeric string id for this app."""
|
|
seen = []
|
|
for coll in ("screens", "widgets", "lines", "groups", "templates"):
|
|
seen.extend(int(k) for k in model.get(coll, {}).keys()
|
|
if str(k).isdigit())
|
|
base = max(seen, default=10000)
|
|
lu = int(model.get("lastUUID") or 10000)
|
|
return max(base + 1, lu + 1)
|
|
|
|
# --------------------------------------------------------------- screens
|
|
def add_screen(self, app_id, name, width=None, height=None):
|
|
model = self.get_app(app_id)
|
|
w = width or model["screenSize"]["w"]
|
|
h = height or model["screenSize"]["h"]
|
|
is_first = len(model.get("screens", {})) == 0
|
|
sid = str(self._next_id(model))
|
|
screen = {
|
|
"id": sid, "name": name, "x": 0, "y": 0, "w": w, "h": h, "z": 0,
|
|
"min": {"h": h, "w": w},
|
|
"props": {"start": is_first},
|
|
"style": {}, "has": {"image": True}, "children": [],
|
|
}
|
|
changes = [
|
|
{"type": "add", "parent": "screens", "name": sid, "object": screen},
|
|
{"type": "update", "name": "lastUUID", "object": int(sid)},
|
|
]
|
|
if is_first:
|
|
changes.append({"type": "update", "name": "startScreen", "object": sid})
|
|
self.apply_changes(app_id, changes)
|
|
return sid
|
|
|
|
# --------------------------------------------------------------- widgets
|
|
def add_widget(self, app_id, screen_id, widget_type, x, y, w, h,
|
|
name=None, props=None, has=None, style=None):
|
|
model = self.get_app(app_id)
|
|
screen = model["screens"].get(screen_id)
|
|
if not screen:
|
|
raise QuantUXError(f"Screen {screen_id} not found in app {app_id}")
|
|
wid = str(self._next_id(model))
|
|
widget = {
|
|
"id": wid,
|
|
"name": name or widget_type,
|
|
"type": widget_type,
|
|
"x": x, "y": y, "w": w, "h": h, "z": 0,
|
|
"props": props or {},
|
|
"has": has if has is not None else _default_has(widget_type),
|
|
"actions": {},
|
|
"style": style if style is not None else _default_style(widget_type),
|
|
}
|
|
screen["children"] = list(screen.get("children", [])) + [wid]
|
|
changes = [
|
|
{"type": "add", "parent": "widgets", "name": wid, "object": widget},
|
|
{"type": "update", "parent": "screens", "name": screen_id,
|
|
"object": screen},
|
|
{"type": "update", "name": "lastUUID", "object": int(wid)},
|
|
]
|
|
self.apply_changes(app_id, changes)
|
|
return wid
|
|
|
|
def update_widget(self, app_id, widget_id, props=None, style=None,
|
|
x=None, y=None, w=None, h=None, name=None):
|
|
model = self.get_app(app_id)
|
|
widget = model["widgets"].get(widget_id)
|
|
if not widget:
|
|
raise QuantUXError(f"Widget {widget_id} not found in app {app_id}")
|
|
if name is not None:
|
|
widget["name"] = name
|
|
if props:
|
|
widget["props"] = {**(widget.get("props") or {}), **props}
|
|
if style:
|
|
widget["style"] = {**(widget.get("style") or {}), **style}
|
|
for key, val in (("x", x), ("y", y), ("w", w), ("h", h)):
|
|
if val is not None:
|
|
widget[key] = val
|
|
return self.apply_changes(app_id, [
|
|
{"type": "update", "parent": "widgets", "name": widget_id,
|
|
"object": widget},
|
|
])
|
|
|
|
def delete_widget(self, app_id, widget_id):
|
|
model = self.get_app(app_id)
|
|
changes = [{"type": "delete", "parent": "widgets", "name": widget_id}]
|
|
for screen in model.get("screens", {}).values():
|
|
if widget_id in screen.get("children", []):
|
|
screen["children"] = [c for c in screen["children"]
|
|
if c != widget_id]
|
|
changes.append({"type": "update", "parent": "screens",
|
|
"name": screen["id"], "object": screen})
|
|
return self.apply_changes(app_id, changes)
|
|
|
|
# ------------------------------------------------------------------ lines
|
|
def connect_flow(self, app_id, from_id, to_id, event="click"):
|
|
"""Wire an interaction: clicking 'from' navigates to 'to'."""
|
|
model = self.get_app(app_id)
|
|
lid = str(self._next_id(model))
|
|
line = {"id": lid, "from": from_id, "to": to_id,
|
|
"event": event, "points": []}
|
|
return self.apply_changes(app_id, [
|
|
{"type": "add", "parent": "lines", "name": lid, "object": line},
|
|
{"type": "update", "name": "lastUUID", "object": int(lid)},
|
|
])
|
|
|
|
# ------------------------------------------------------------------ misc
|
|
def describe(self, app_id):
|
|
"""Human/agent friendly summary of an app model."""
|
|
m = self.get_app(app_id)
|
|
out = {
|
|
"id": m.get("_id") or m.get("id"),
|
|
"name": m.get("name"),
|
|
"type": m.get("type"),
|
|
"screenSize": m.get("screenSize"),
|
|
"startScreen": m.get("startScreen"),
|
|
"screens": [],
|
|
"widgets": len(m.get("widgets", {})),
|
|
"lines": len(m.get("lines", {})),
|
|
}
|
|
for sid, s in (m.get("screens") or {}).items():
|
|
out["screens"].append({
|
|
"id": s.get("id"), "name": s.get("name"),
|
|
"w": s.get("w"), "h": s.get("h"),
|
|
"start": (s.get("props") or {}).get("start", False),
|
|
"widgetCount": len(s.get("children") or []),
|
|
})
|
|
return out
|
|
|
|
# ------------------------------------------------------------- transport
|
|
def _req(self, method, path, **kw):
|
|
allow_auth_retry = kw.pop("_allow_auth_retry", True)
|
|
url = self.base_url + path
|
|
kw.setdefault("timeout", self.timeout)
|
|
try:
|
|
resp = self.session.request(method, url, **kw)
|
|
except requests.RequestException as exc:
|
|
raise QuantUXError(f"Request to {url} failed: {exc}") from exc
|
|
if (
|
|
resp.status_code in (401, 403)
|
|
and allow_auth_retry
|
|
and self._auth_refresh_handler is not None
|
|
):
|
|
failed_token = self.token
|
|
self._auth_refresh_handler(failed_token)
|
|
return self._req(method, path, _allow_auth_retry=False, **kw)
|
|
if resp.status_code >= 400:
|
|
body = resp.text[:500]
|
|
raise QuantUXError(f"HTTP {resp.status_code} from {method} {path}: {body}")
|
|
try:
|
|
return resp.json()
|
|
except ValueError:
|
|
return resp.text
|