307 lines
10 KiB
Python
307 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
"""Core configuration REST API (buttons, layout, background, feedback,
|
|
settings, icons) for the ButtonTask web configurator."""
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
from flask import Blueprint, abort, jsonify, request, send_from_directory
|
|
|
|
from auth import login_required
|
|
from config_store import (ALLOWED_ICON_EXT, icons_dir, list_icons, load_config,
|
|
save_config)
|
|
|
|
api_bp = Blueprint("api", __name__)
|
|
|
|
_RESPONSE_RESULTS = {"ok", "error", "pending"}
|
|
|
|
|
|
def _validate_response_check(data):
|
|
if data is None:
|
|
return None
|
|
if not isinstance(data, dict):
|
|
abort(400, "responseCheck must be an object")
|
|
|
|
enabled = bool(data.get("enabled", False))
|
|
if not enabled:
|
|
return {"enabled": False}
|
|
|
|
rules = data.get("rules", [])
|
|
if not isinstance(rules, list):
|
|
abort(400, "responseCheck.rules must be an array")
|
|
|
|
normalized_rules = []
|
|
for rule in rules:
|
|
if not isinstance(rule, dict):
|
|
abort(400, "each responseCheck rule must be an object")
|
|
field = (rule.get("field") or "").strip()
|
|
if not field:
|
|
abort(400, "responseCheck rule field is required")
|
|
result = (rule.get("result") or "error").strip().lower()
|
|
if result not in _RESPONSE_RESULTS:
|
|
abort(400, "responseCheck rule result must be ok, error or pending")
|
|
if not rule.get("equals") and rule.get("contains") in (None, ""):
|
|
abort(400, "responseCheck rule needs equals or contains")
|
|
item = {"field": field, "result": result}
|
|
if rule.get("equals") not in (None, ""):
|
|
item["equals"] = str(rule.get("equals"))
|
|
if rule.get("contains") not in (None, ""):
|
|
item["contains"] = str(rule.get("contains"))
|
|
message = (rule.get("message") or "").strip()
|
|
if message:
|
|
item["message"] = message
|
|
normalized_rules.append(item)
|
|
|
|
default_result = (data.get("defaultResult") or "error").strip().lower()
|
|
if default_result not in _RESPONSE_RESULTS:
|
|
abort(400, "defaultResult must be ok, error or pending")
|
|
|
|
poll_in = data.get("poll") or {}
|
|
if poll_in and not isinstance(poll_in, dict):
|
|
abort(400, "responseCheck.poll must be an object")
|
|
interval_ms = int(poll_in.get("intervalMs", 3000))
|
|
max_attempts = int(poll_in.get("maxAttempts", 20))
|
|
timeout_ms = int(poll_in.get("timeoutMs", 120000))
|
|
if interval_ms < 500 or interval_ms > 60000:
|
|
abort(400, "poll.intervalMs must be 500..60000")
|
|
if max_attempts < 1 or max_attempts > 100:
|
|
abort(400, "poll.maxAttempts must be 1..100")
|
|
if timeout_ms < interval_ms or timeout_ms > 600000:
|
|
abort(400, "poll.timeoutMs must be >= intervalMs and <= 600000")
|
|
|
|
return {
|
|
"enabled": True,
|
|
"requireHttpSuccess": bool(data.get("requireHttpSuccess", True)),
|
|
"rules": normalized_rules,
|
|
"defaultResult": default_result,
|
|
"poll": {
|
|
"intervalMs": interval_ms,
|
|
"maxAttempts": max_attempts,
|
|
"timeoutMs": timeout_ms,
|
|
},
|
|
}
|
|
|
|
|
|
def _apply_button_action(action, data):
|
|
action["type"] = data.get("actionType", action.get("type", "http_get"))
|
|
action["url"] = data.get("url", action.get("url", ""))
|
|
if "headers" in data:
|
|
action["headers"] = data["headers"]
|
|
if "body" in data:
|
|
action["body"] = data["body"]
|
|
if "responseCheck" in data:
|
|
rc = _validate_response_check(data.get("responseCheck"))
|
|
if rc is None:
|
|
action.pop("responseCheck", None)
|
|
else:
|
|
action["responseCheck"] = rc
|
|
return action
|
|
|
|
|
|
@api_bp.route("/icons/<path:fname>")
|
|
@login_required
|
|
def serve_icon(fname):
|
|
return send_from_directory(str(icons_dir()), fname)
|
|
|
|
|
|
# --- config ---
|
|
@api_bp.route("/api/config", methods=["GET"])
|
|
@login_required
|
|
def api_get_config():
|
|
return jsonify(load_config())
|
|
|
|
|
|
@api_bp.route("/api/config", methods=["PUT"])
|
|
@login_required
|
|
def api_put_config():
|
|
data = request.get_json(force=True, silent=True)
|
|
if not isinstance(data, dict):
|
|
abort(400, "JSON object required")
|
|
save_config(data)
|
|
return jsonify({"ok": True})
|
|
|
|
|
|
# --- buttons ---
|
|
@api_bp.route("/api/buttons", methods=["POST"])
|
|
@login_required
|
|
def api_add_button():
|
|
data = request.get_json(force=True, silent=True) or {}
|
|
cfg = load_config()
|
|
btn = {
|
|
"id": data.get("id") or uuid.uuid4().hex,
|
|
"label": data.get("label", "Кнопка"),
|
|
"iconPath": data.get("iconPath", ""),
|
|
"action": _apply_button_action({
|
|
"type": data.get("actionType", "http_get"),
|
|
"url": data.get("url", ""),
|
|
"headers": data.get("headers", {}),
|
|
"body": data.get("body", ""),
|
|
}, data),
|
|
"feedback": {
|
|
"successText": data.get("successText", ""),
|
|
"errorText": data.get("errorText", "Ошибка"),
|
|
"pendingText": data.get("pendingText", "..."),
|
|
"fadeMs": int(data.get("fadeMs", 5000)),
|
|
},
|
|
"trigger": {
|
|
"mode": "click" if data.get("triggerMode") == "click" else "hold",
|
|
"holdMs": int(data.get("holdMs", 800)),
|
|
},
|
|
"color": data.get("color", "#7b007b"),
|
|
}
|
|
cfg.setdefault("buttons", []).append(btn)
|
|
save_config(cfg)
|
|
return jsonify(btn)
|
|
|
|
|
|
@api_bp.route("/api/buttons/<bid>", methods=["PUT"])
|
|
@login_required
|
|
def api_update_button(bid):
|
|
data = request.get_json(force=True, silent=True) or {}
|
|
cfg = load_config()
|
|
for b in cfg.get("buttons", []):
|
|
if b.get("id") == bid:
|
|
b["label"] = data.get("label", b.get("label"))
|
|
b["iconPath"] = data.get("iconPath", b.get("iconPath"))
|
|
action = b.setdefault("action", {})
|
|
_apply_button_action(action, data)
|
|
feedback = b.setdefault("feedback", {})
|
|
feedback["successText"] = data.get("successText", feedback.get("successText", ""))
|
|
feedback["errorText"] = data.get("errorText", feedback.get("errorText", "Ошибка"))
|
|
feedback["pendingText"] = data.get("pendingText", feedback.get("pendingText", "..."))
|
|
feedback["fadeMs"] = int(data.get("fadeMs", feedback.get("fadeMs", 5000)))
|
|
trigger = b.setdefault("trigger", {})
|
|
if "triggerMode" in data:
|
|
trigger["mode"] = "click" if data["triggerMode"] == "click" else "hold"
|
|
else:
|
|
trigger.setdefault("mode", "hold")
|
|
if "holdMs" in data:
|
|
trigger["holdMs"] = int(data["holdMs"])
|
|
else:
|
|
trigger.setdefault("holdMs", 800)
|
|
if "color" in data:
|
|
b["color"] = data["color"]
|
|
save_config(cfg)
|
|
return jsonify(b)
|
|
abort(404)
|
|
|
|
|
|
@api_bp.route("/api/buttons/<bid>", methods=["DELETE"])
|
|
@login_required
|
|
def api_delete_button(bid):
|
|
cfg = load_config()
|
|
cfg["buttons"] = [b for b in cfg.get("buttons", []) if b.get("id") != bid]
|
|
save_config(cfg)
|
|
return jsonify({"ok": True})
|
|
|
|
|
|
@api_bp.route("/api/buttons/reorder", methods=["POST"])
|
|
@login_required
|
|
def api_reorder_buttons():
|
|
data = request.get_json(force=True, silent=True) or {}
|
|
order = data.get("order", [])
|
|
cfg = load_config()
|
|
by_id = {b.get("id"): b for b in cfg.get("buttons", [])}
|
|
new_list = [by_id[i] for i in order if i in by_id]
|
|
for b in cfg.get("buttons", []):
|
|
if b not in new_list:
|
|
new_list.append(b)
|
|
cfg["buttons"] = new_list
|
|
save_config(cfg)
|
|
return jsonify({"ok": True, "count": len(new_list)})
|
|
|
|
|
|
# --- layout / background / feedback / settings ---
|
|
@api_bp.route("/api/layout", methods=["PUT"])
|
|
@login_required
|
|
def api_set_layout():
|
|
data = request.get_json(force=True, silent=True) or {}
|
|
cfg = load_config()
|
|
layout = cfg.setdefault("layout", {})
|
|
for k in ("mode", "columns", "spacing", "showLabels"):
|
|
if k in data:
|
|
layout[k] = data[k]
|
|
save_config(cfg)
|
|
return jsonify(layout)
|
|
|
|
|
|
@api_bp.route("/api/background", methods=["PUT"])
|
|
@login_required
|
|
def api_set_background():
|
|
data = request.get_json(force=True, silent=True) or {}
|
|
cfg = load_config()
|
|
bg = cfg.setdefault("background", {})
|
|
for k in ("type", "color1", "color2", "animated", "imagePath"):
|
|
if k in data:
|
|
bg[k] = data[k]
|
|
save_config(cfg)
|
|
return jsonify(bg)
|
|
|
|
|
|
@api_bp.route("/api/feedback", methods=["PUT"])
|
|
@login_required
|
|
def api_set_feedback():
|
|
data = request.get_json(force=True, silent=True) or {}
|
|
cfg = load_config()
|
|
fb = cfg.setdefault("feedback", {})
|
|
for k in ("okColor", "errorColor"):
|
|
if k in data:
|
|
fb[k] = data[k]
|
|
for k in ("okWidth", "errorWidth", "glowRadius"):
|
|
if k in data:
|
|
try:
|
|
fb[k] = int(data[k])
|
|
except (TypeError, ValueError):
|
|
pass
|
|
save_config(cfg)
|
|
return jsonify(fb)
|
|
|
|
|
|
@api_bp.route("/api/settings", methods=["PUT"])
|
|
@login_required
|
|
def api_set_settings():
|
|
data = request.get_json(force=True, silent=True) or {}
|
|
cfg = load_config()
|
|
st = cfg.setdefault("settings", {})
|
|
for k in ("darkMode", "password", "kioskMode", "iconsDir", "webPort"):
|
|
if k in data:
|
|
st[k] = data[k]
|
|
save_config(cfg)
|
|
return jsonify({"ok": True})
|
|
|
|
|
|
# --- icons ---
|
|
@api_bp.route("/api/icons", methods=["GET"])
|
|
@login_required
|
|
def api_list_icons():
|
|
return jsonify(list_icons())
|
|
|
|
|
|
@api_bp.route("/api/icons", methods=["POST"])
|
|
@login_required
|
|
def api_upload_icon():
|
|
if "file" not in request.files:
|
|
abort(400, "file required")
|
|
fobj = request.files["file"]
|
|
fname = Path(fobj.filename).name
|
|
if not fname:
|
|
abort(400, "empty filename")
|
|
ext = Path(fname).suffix.lower()
|
|
if ext not in ALLOWED_ICON_EXT:
|
|
abort(400, f"extension {ext} not allowed")
|
|
target = icons_dir() / fname
|
|
fobj.save(str(target))
|
|
return jsonify({"ok": True, "name": fname})
|
|
|
|
|
|
@api_bp.route("/api/icons/<path:fname>", methods=["DELETE"])
|
|
@login_required
|
|
def api_delete_icon(fname):
|
|
p = icons_dir() / fname
|
|
try:
|
|
p.resolve().relative_to(icons_dir().resolve())
|
|
except ValueError:
|
|
abort(400)
|
|
if p.exists():
|
|
p.unlink()
|
|
return jsonify({"ok": True})
|