374 lines
12 KiB
Python
374 lines
12 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 _validate_latch_reset(data):
|
||
if data is None:
|
||
return None
|
||
if not isinstance(data, dict):
|
||
abort(400, "latchReset must be an object")
|
||
|
||
enabled = bool(data.get("enabled", False))
|
||
if not enabled:
|
||
return {"enabled": False}
|
||
|
||
reset_url = (data.get("resetUrl") or "").strip()
|
||
if not reset_url:
|
||
abort(400, "latchReset.resetUrl is required when enabled")
|
||
success_match = (data.get("successMatch") or "OK").strip() or "OK"
|
||
result = {
|
||
"enabled": True,
|
||
"resetUrl": reset_url,
|
||
"successMatch": success_match,
|
||
}
|
||
if data.get("fireAndForget"):
|
||
result["fireAndForget"] = True
|
||
return result
|
||
|
||
|
||
def _apply_button_latch_reset(btn, data):
|
||
if "latchReset" not in data:
|
||
return btn
|
||
lr = _validate_latch_reset(data.get("latchReset"))
|
||
if lr is None:
|
||
btn.pop("latchReset", None)
|
||
else:
|
||
btn["latchReset"] = lr
|
||
return btn
|
||
|
||
|
||
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 "timeoutMs" in data:
|
||
try:
|
||
timeout_ms = int(data["timeoutMs"])
|
||
except (TypeError, ValueError):
|
||
abort(400, "timeoutMs must be an integer")
|
||
if timeout_ms < 1000 or timeout_ms > 600000:
|
||
abort(400, "timeoutMs must be between 1000 and 600000")
|
||
if timeout_ms == 7000:
|
||
action.pop("timeoutMs", None)
|
||
else:
|
||
action["timeoutMs"] = timeout_ms
|
||
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"),
|
||
}
|
||
_apply_button_latch_reset(btn, data)
|
||
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"]
|
||
_apply_button_latch_reset(b, data)
|
||
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", "labelColor"):
|
||
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", "holdColor", "pendingColor"):
|
||
if k in data:
|
||
fb[k] = data[k]
|
||
for k in ("okWidth", "errorWidth", "glowRadius", "pendingWidth"):
|
||
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 {}
|
||
nested = data.get("settings")
|
||
if isinstance(nested, dict):
|
||
data = {**data, **nested}
|
||
cfg = load_config()
|
||
st = cfg.setdefault("settings", {})
|
||
if not isinstance(st, dict):
|
||
st = {}
|
||
for k in ("darkMode", "password", "kioskMode", "iconsDir", "webPort"):
|
||
if k not in data:
|
||
continue
|
||
val = data[k]
|
||
if k == "password":
|
||
if not isinstance(val, str) or not val.strip():
|
||
abort(400, "password required")
|
||
val = val.strip()
|
||
if k == "webPort":
|
||
try:
|
||
val = int(val)
|
||
except (TypeError, ValueError):
|
||
abort(400, "invalid web port")
|
||
if val < 1024 or val > 65535:
|
||
abort(400, "web port must be 1024–65535")
|
||
st[k] = val
|
||
cfg["settings"] = st
|
||
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})
|