151 lines
5.2 KiB
Python
151 lines
5.2 KiB
Python
from flask import current_app, flash, jsonify, redirect, render_template, request, url_for
|
|
from flask_login import current_user, login_required, login_user, logout_user
|
|
|
|
from .core import LicenseService, RuntimeState, User, discover_node, get_status_info, validate_node_payload
|
|
|
|
|
|
def get_runtime_state() -> RuntimeState:
|
|
return current_app.extensions["runtime_state"]
|
|
|
|
|
|
def get_license_service() -> LicenseService:
|
|
return current_app.extensions["license_service"]
|
|
|
|
|
|
def json_error(message: str, status_code: int = 400):
|
|
return jsonify({"status": "error", "message": message}), status_code
|
|
|
|
|
|
def register_routes(app):
|
|
@app.route("/login", methods=["GET", "POST"])
|
|
def login():
|
|
if current_user.is_authenticated:
|
|
return redirect(url_for("index"))
|
|
|
|
if request.method == "POST":
|
|
username = request.form.get("username", "").strip()
|
|
password = request.form.get("password", "")
|
|
if get_runtime_state().authenticate(username, password):
|
|
login_user(User(username))
|
|
return redirect(url_for("index"))
|
|
flash("Usuario ou senha invalidos.")
|
|
return render_template("login.html")
|
|
|
|
@app.route("/")
|
|
@login_required
|
|
def index():
|
|
config, _, _, _ = get_runtime_state().snapshot()
|
|
license_status = get_license_service().verify()
|
|
return render_template(
|
|
"index.html",
|
|
config=config,
|
|
licenciado=license_status.valid,
|
|
lic_msg=license_status.message,
|
|
hwid=license_status.hardware_id,
|
|
dias_restantes=license_status.days_remaining,
|
|
)
|
|
|
|
@app.route("/api/nodes", methods=["POST"])
|
|
@login_required
|
|
def api_nodes():
|
|
payload = request.get_json(silent=True)
|
|
if not isinstance(payload, dict):
|
|
return json_error("JSON invalido.")
|
|
|
|
try:
|
|
zabbix_server = str(payload.get("zabbix_server", "")).strip()
|
|
hostname_zabbix = str(payload.get("hostname_zabbix", "")).strip()
|
|
nodes = [validate_node_payload(node) for node in payload.get("nodes", [])]
|
|
except ValueError as exc:
|
|
return json_error(str(exc))
|
|
|
|
if not zabbix_server:
|
|
return json_error("Zabbix Server e obrigatorio.")
|
|
if not hostname_zabbix:
|
|
return json_error("Hostname Zabbix e obrigatorio.")
|
|
|
|
get_runtime_state().update_config(zabbix_server, hostname_zabbix, nodes)
|
|
return jsonify({"status": "success"})
|
|
|
|
@app.route("/api/rename", methods=["POST"])
|
|
@login_required
|
|
def api_rename():
|
|
payload = request.get_json(silent=True)
|
|
if not isinstance(payload, dict):
|
|
return json_error("JSON invalido.")
|
|
|
|
address_key = str(payload.get("addr", "")).strip()
|
|
label = str(payload.get("label", "")).strip()
|
|
|
|
if not address_key:
|
|
return json_error("Endereco do dispositivo e obrigatorio.")
|
|
if not label:
|
|
return json_error("Nome do dispositivo e obrigatorio.")
|
|
|
|
if not get_runtime_state().rename_device(address_key, label):
|
|
return json_error("Dispositivo nao encontrado.", 404)
|
|
|
|
return jsonify({"status": "success"})
|
|
|
|
@app.route("/api/license", methods=["POST"])
|
|
def api_license():
|
|
payload = request.get_json(silent=True)
|
|
if not isinstance(payload, dict):
|
|
return json_error("JSON invalido.")
|
|
|
|
try:
|
|
get_license_service().save(str(payload.get("serial", "")))
|
|
except ValueError as exc:
|
|
return json_error(str(exc))
|
|
|
|
get_runtime_state().trigger_restart()
|
|
return jsonify({"status": "success"})
|
|
|
|
@app.route("/api/rescan_node", methods=["POST"])
|
|
@login_required
|
|
def api_rescan():
|
|
payload = request.get_json(silent=True)
|
|
if not isinstance(payload, dict):
|
|
return json_error("JSON invalido.")
|
|
|
|
try:
|
|
node = validate_node_payload(payload)
|
|
except ValueError as exc:
|
|
return json_error(str(exc))
|
|
|
|
_, current_map, _, _ = get_runtime_state().snapshot()
|
|
discovered_map = discover_node(node, current_map)
|
|
get_runtime_state().replace_node_devices(node, discovered_map)
|
|
return jsonify({"status": "success", "count": len(discovered_map)})
|
|
|
|
@app.route("/api/data")
|
|
@login_required
|
|
def api_data():
|
|
_, device_map, realtime_data, node_statuses = get_runtime_state().snapshot()
|
|
sensors = []
|
|
|
|
for key, info in device_map.items():
|
|
data = realtime_data.get(
|
|
key,
|
|
{"status": 0, "last": "--:--:--", "node": info.get("node_nome", "...")},
|
|
)
|
|
status_info = get_status_info(int(data["status"]))
|
|
sensors.append(
|
|
{
|
|
"key": key,
|
|
"label": info["label"],
|
|
"node": data.get("node", "..."),
|
|
"status_text": status_info["label"],
|
|
"color": status_info["color"],
|
|
"last": data["last"],
|
|
}
|
|
)
|
|
|
|
return jsonify({"sensores": sensors, "nodes_status": node_statuses})
|
|
|
|
@app.route("/logout")
|
|
@login_required
|
|
def logout():
|
|
logout_user()
|
|
return redirect(url_for("login"))
|