443 lines
14 KiB
Python
443 lines
14 KiB
Python
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
from flask import Flask, Response, jsonify, request, send_from_directory
|
|
from flask_cors import CORS
|
|
|
|
from db_py import count_tests, del_config, get_all_tests, get_config, set_config
|
|
from scanner import full_scan
|
|
from sse_py import broadcast, stream_events
|
|
from watcher import start_watching
|
|
|
|
PORT = int(os.getenv("PORT", "3001"))
|
|
ALLOWED_KEYS = {
|
|
"target_dir",
|
|
"results_dir",
|
|
"avg_time_coe",
|
|
"avg_time_p2p",
|
|
"avg_time_p3p",
|
|
}
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
DIST_DIR = BASE_DIR.parent / "dashboard" / "dist"
|
|
|
|
app = Flask(__name__, static_folder=str(DIST_DIR), static_url_path="")
|
|
CORS(app)
|
|
|
|
|
|
@app.get("/api/tests")
|
|
def get_tests_route():
|
|
completed = request.args.get("completed")
|
|
interference = request.args.get("interference")
|
|
throttled = request.args.get("throttled")
|
|
device = request.args.get("device")
|
|
rotation = request.args.get("rotation")
|
|
test_point = request.args.get("testPoint")
|
|
station = request.args.get("station")
|
|
band = request.args.get("band")
|
|
channel = request.args.get("channel")
|
|
bandwidth = request.args.get("bandwidth")
|
|
rssi = request.args.get("rssi")
|
|
direction = request.args.get("direction")
|
|
|
|
tests = get_all_tests()
|
|
|
|
if completed is not None:
|
|
completed_value = 1 if completed == "true" else 0
|
|
tests = [t for t in tests if t.get("completed") == completed_value]
|
|
if interference:
|
|
tests = [t for t in tests if t.get("interference") == interference]
|
|
if throttled:
|
|
tests = [t for t in tests if t.get("throttled") == throttled]
|
|
if device:
|
|
tests = [t for t in tests if t.get("device") == device]
|
|
if rotation:
|
|
tests = [t for t in tests if t.get("rotation") == rotation]
|
|
if test_point:
|
|
tests = [t for t in tests if t.get("test_point") == test_point]
|
|
if station:
|
|
tests = [t for t in tests if t.get("station") == station]
|
|
if band:
|
|
tests = [t for t in tests if t.get("band") == band]
|
|
if channel:
|
|
tests = [t for t in tests if t.get("channel") == channel]
|
|
if bandwidth:
|
|
tests = [t for t in tests if t.get("bandwidth") == bandwidth]
|
|
if rssi:
|
|
tests = [t for t in tests if t.get("rssi") == rssi]
|
|
if direction:
|
|
tests = [t for t in tests if t.get("direction") == direction]
|
|
|
|
tests.sort(key=lambda item: ((item.get("interference") or ""), (item.get("test_id") or "")))
|
|
return jsonify(tests)
|
|
|
|
|
|
@app.get("/api/stats")
|
|
def get_stats_route():
|
|
tests = get_all_tests()
|
|
types = ["COE", "P2P", "P3P"]
|
|
|
|
total_completed = sum(1 for t in tests if t.get("completed"))
|
|
overall = {
|
|
"total": len(tests),
|
|
"completed": total_completed,
|
|
"completionRate": (total_completed / len(tests)) if tests else 0,
|
|
}
|
|
|
|
device_map = {}
|
|
for test in tests:
|
|
name = test.get("device")
|
|
if not name:
|
|
continue
|
|
|
|
if name not in device_map:
|
|
device_map[name] = {
|
|
"total": 0,
|
|
"completed": 0,
|
|
"byType": {t: {"total": 0, "completed": 0} for t in types},
|
|
}
|
|
device_map[name]["total"] += 1
|
|
|
|
interference = test.get("interference")
|
|
if interference in device_map[name]["byType"]:
|
|
device_map[name]["byType"][interference]["total"] += 1
|
|
|
|
if test.get("completed"):
|
|
device_map[name]["completed"] += 1
|
|
if interference in device_map[name]["byType"]:
|
|
device_map[name]["byType"][interference]["completed"] += 1
|
|
|
|
devices = []
|
|
for name in sorted(device_map.keys()):
|
|
stats = device_map[name]
|
|
devices.append(
|
|
{
|
|
"name": name,
|
|
"total": stats["total"],
|
|
"completed": stats["completed"],
|
|
"completionRate": (stats["completed"] / stats["total"]) if stats["total"] else 0,
|
|
"byType": stats["byType"],
|
|
}
|
|
)
|
|
|
|
elapsed_seconds = 0
|
|
for test in tests:
|
|
duration = test.get("duration_seconds")
|
|
if test.get("completed") and duration is not None:
|
|
elapsed_seconds += duration
|
|
|
|
by_type = {}
|
|
estimate_possible = True
|
|
estimated_remaining_seconds = 0
|
|
|
|
for test_type in types:
|
|
type_tests = [t for t in tests if t.get("interference") == test_type]
|
|
completed_tests = [t for t in type_tests if t.get("completed")]
|
|
with_duration = [t for t in completed_tests if t.get("duration_seconds") is not None]
|
|
remaining = len(type_tests) - len(completed_tests)
|
|
|
|
calc_avg = None
|
|
if with_duration:
|
|
calc_avg = sum(t.get("duration_seconds") for t in with_duration) / len(with_duration)
|
|
|
|
manual_value = get_config(f"avg_time_{test_type.lower()}")
|
|
|
|
avg_seconds = None
|
|
avg_source = None
|
|
|
|
if manual_value is not None:
|
|
avg_seconds = float(manual_value)
|
|
avg_source = "manual_override" if calc_avg is not None else "manual"
|
|
elif calc_avg is not None:
|
|
avg_seconds = calc_avg
|
|
avg_source = "calculated"
|
|
|
|
by_type[test_type] = {
|
|
"total": len(type_tests),
|
|
"completed": len(completed_tests),
|
|
"remaining": remaining,
|
|
"avgSeconds": avg_seconds,
|
|
"avgSource": avg_source,
|
|
}
|
|
|
|
if remaining > 0:
|
|
if avg_seconds is not None:
|
|
estimated_remaining_seconds += avg_seconds * remaining
|
|
else:
|
|
estimate_possible = False
|
|
|
|
return jsonify(
|
|
{
|
|
"overall": overall,
|
|
"devices": devices,
|
|
"timing": {
|
|
"elapsedSeconds": elapsed_seconds,
|
|
"estimatedRemainingSeconds": estimated_remaining_seconds if estimate_possible else None,
|
|
"byType": by_type,
|
|
},
|
|
}
|
|
)
|
|
|
|
|
|
@app.get("/api/config")
|
|
def get_config_route():
|
|
config = {}
|
|
for key in ALLOWED_KEYS:
|
|
config[key] = get_config(key)
|
|
return jsonify(config)
|
|
|
|
|
|
@app.post("/api/config")
|
|
def set_config_route():
|
|
updates = request.get_json(silent=True)
|
|
if not isinstance(updates, dict):
|
|
return jsonify({"error": "Request body must be a JSON object"}), 400
|
|
|
|
dirs_changed = False
|
|
|
|
for key, value in updates.items():
|
|
if key not in ALLOWED_KEYS:
|
|
continue
|
|
|
|
if value in (None, ""):
|
|
del_config(key)
|
|
else:
|
|
set_config(key, str(value))
|
|
|
|
if key in {"target_dir", "results_dir"}:
|
|
dirs_changed = True
|
|
|
|
if dirs_changed:
|
|
target_dir = get_config("target_dir")
|
|
results_dir = get_config("results_dir")
|
|
try:
|
|
full_scan(target_dir, results_dir)
|
|
start_watching(target_dir, results_dir)
|
|
except Exception as exc:
|
|
print(f"[config] fullScan error: {exc}")
|
|
return jsonify({"error": f"Scan failed: {exc}"}), 500
|
|
|
|
tests = get_all_tests()
|
|
completed = len([t for t in tests if t.get("completed")])
|
|
print(f"[config] Scan complete -> {len(tests)} tests found, {completed} completed")
|
|
broadcast({"type": "update"})
|
|
return jsonify({"ok": True, "testCount": len(tests), "completedCount": completed})
|
|
|
|
return jsonify({"ok": True, "testCount": None, "completedCount": None})
|
|
|
|
|
|
@app.post("/api/config/rescan")
|
|
def rescan_route():
|
|
target_dir = get_config("target_dir")
|
|
results_dir = get_config("results_dir")
|
|
if not target_dir or not results_dir:
|
|
return jsonify({"error": "Directories not configured"}), 400
|
|
|
|
try:
|
|
full_scan(target_dir, results_dir)
|
|
start_watching(target_dir, results_dir)
|
|
except Exception as exc:
|
|
print(f"[config] rescan error: {exc}")
|
|
return jsonify({"error": f"Scan failed: {exc}"}), 500
|
|
|
|
tests = get_all_tests()
|
|
completed = len([t for t in tests if t.get("completed")])
|
|
print(f"[config] Rescan complete -> {len(tests)} tests, {completed} completed")
|
|
broadcast({"type": "update"})
|
|
return jsonify({"ok": True, "testCount": len(tests), "completedCount": completed})
|
|
|
|
|
|
@app.get("/api/browse")
|
|
def browse_route():
|
|
def _normalize_request_path(path):
|
|
if not path:
|
|
return path
|
|
p = path.strip()
|
|
if os.name == "nt":
|
|
p = p.replace("/", "\\")
|
|
return p
|
|
|
|
def _is_unc_host_only(path):
|
|
if os.name != "nt" or not path:
|
|
return False
|
|
p = path.rstrip("\\")
|
|
if not p.startswith("\\\\"):
|
|
return False
|
|
remainder = p[2:]
|
|
return bool(remainder) and "\\" not in remainder
|
|
|
|
def _shares_for_unc_host(host):
|
|
# Enumerate SMB shares with native Windows tooling for host-only UNC paths.
|
|
proc = subprocess.run(
|
|
["net", "view", f"\\\\{host}"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
check=False,
|
|
)
|
|
if proc.returncode != 0:
|
|
raise RuntimeError(proc.stderr.strip() or proc.stdout.strip() or "Cannot query network host")
|
|
|
|
shares = []
|
|
in_table = False
|
|
for raw_line in proc.stdout.splitlines():
|
|
line = raw_line.strip()
|
|
if not line:
|
|
continue
|
|
if line.startswith("---"):
|
|
in_table = True
|
|
continue
|
|
if not in_table:
|
|
continue
|
|
if line.lower().startswith("the command completed successfully"):
|
|
break
|
|
|
|
first = line.split()[0]
|
|
if first and first.lower() not in {"share", "name"}:
|
|
shares.append(first)
|
|
|
|
unique = sorted(set(shares), key=str.lower)
|
|
return [{"name": share, "path": f"\\\\{host}\\{share}"} for share in unique]
|
|
|
|
def _configured_roots():
|
|
raw = os.getenv("BROWSE_ROOTS", "").strip()
|
|
if not raw:
|
|
return []
|
|
roots = []
|
|
for part in raw.split(";"):
|
|
p = part.strip()
|
|
if not p:
|
|
continue
|
|
abs_p = os.path.abspath(p)
|
|
if os.path.isdir(abs_p):
|
|
roots.append(abs_p)
|
|
return roots
|
|
|
|
def _is_within_allowed_roots(path, roots):
|
|
if not roots:
|
|
return True
|
|
normalized = os.path.normcase(os.path.abspath(path))
|
|
for root in roots:
|
|
try:
|
|
if os.path.commonpath([normalized, root]) == root:
|
|
return True
|
|
except ValueError:
|
|
continue
|
|
return False
|
|
|
|
def _is_unc_host_allowed(host, roots):
|
|
if not roots:
|
|
return True
|
|
host_prefix = os.path.normcase(f"\\\\{host}\\")
|
|
host_exact = host_prefix.rstrip("\\")
|
|
for root in roots:
|
|
normalized_root = os.path.normcase(root)
|
|
if normalized_root == host_exact or normalized_root.startswith(host_prefix):
|
|
return True
|
|
return False
|
|
|
|
roots = _configured_roots()
|
|
req_path = _normalize_request_path(request.args.get("path"))
|
|
|
|
if not req_path:
|
|
if roots:
|
|
dirs = []
|
|
for root in roots:
|
|
display_name = os.path.basename(root.rstrip("/\\")) or root
|
|
dirs.append({"name": display_name, "path": root})
|
|
return jsonify({"path": None, "parent": None, "dirs": dirs})
|
|
|
|
if os.name == "nt":
|
|
dirs = []
|
|
for drive_idx in range(65, 91):
|
|
drive = f"{chr(drive_idx)}:\\"
|
|
if os.path.exists(drive):
|
|
dirs.append({"name": drive, "path": drive})
|
|
else:
|
|
dirs = [{"name": "/", "path": "/"}]
|
|
|
|
return jsonify({"path": None, "parent": None, "dirs": dirs})
|
|
|
|
if _is_unc_host_only(req_path):
|
|
host = req_path.rstrip("\\")[2:]
|
|
if not _is_unc_host_allowed(host, roots):
|
|
return jsonify({"error": "Path is outside allowed browse roots"}), 403
|
|
try:
|
|
dirs = _shares_for_unc_host(host)
|
|
except Exception as exc:
|
|
return jsonify({"error": f"Cannot list shares for host: {exc}"}), 403
|
|
return jsonify({"path": f"\\\\{host}", "parent": None, "dirs": dirs})
|
|
|
|
if not os.path.exists(req_path):
|
|
return jsonify({"error": "Path does not exist"}), 400
|
|
if not os.path.isdir(req_path):
|
|
return jsonify({"error": "Path is not a directory"}), 400
|
|
if not _is_within_allowed_roots(req_path, roots):
|
|
return jsonify({"error": "Path is outside allowed browse roots"}), 403
|
|
|
|
try:
|
|
dirs = []
|
|
for name in os.listdir(req_path):
|
|
child = os.path.join(req_path, name)
|
|
if os.path.isdir(child) and not name.startswith("."):
|
|
dirs.append({"name": name, "path": child})
|
|
|
|
dirs.sort(key=lambda item: item["name"].lower())
|
|
except OSError:
|
|
return jsonify({"error": "Cannot read directory"}), 403
|
|
|
|
parent = os.path.dirname(req_path)
|
|
at_root = os.path.normcase(parent) == os.path.normcase(req_path)
|
|
if roots and parent and not _is_within_allowed_roots(parent, roots):
|
|
parent = None
|
|
return jsonify({"path": req_path, "parent": None if at_root else parent, "dirs": dirs})
|
|
|
|
|
|
@app.get("/api/events")
|
|
def events_route():
|
|
headers = {
|
|
"Cache-Control": "no-cache",
|
|
"Connection": "keep-alive",
|
|
"X-Accel-Buffering": "no",
|
|
}
|
|
return Response(stream_events(), mimetype="text/event-stream", headers=headers)
|
|
|
|
|
|
@app.get("/")
|
|
@app.get("/<path:path>")
|
|
def static_or_spa(path=""):
|
|
if not DIST_DIR.exists():
|
|
return jsonify({"error": "Frontend dist not found"}), 404
|
|
|
|
if path and (DIST_DIR / path).is_file():
|
|
return send_from_directory(DIST_DIR, path)
|
|
return send_from_directory(DIST_DIR, "index.html")
|
|
|
|
|
|
def bootstrap():
|
|
target_dir = get_config("target_dir")
|
|
results_dir = get_config("results_dir")
|
|
|
|
if target_dir and results_dir:
|
|
existing = count_tests()
|
|
if existing > 0:
|
|
print(f"[server] Resuming from DB -> {existing} tests already loaded.")
|
|
else:
|
|
print("[server] No cached data, scanning directories...")
|
|
full_scan(target_dir, results_dir)
|
|
tests = get_all_tests()
|
|
completed = len([t for t in tests if t.get("completed")])
|
|
print(f"[server] Scan complete -> {len(tests)} tests found, {completed} completed")
|
|
|
|
start_watching(target_dir, results_dir)
|
|
print("[server] Watching for changes.")
|
|
else:
|
|
print("[server] No directories configured -> open the dashboard settings to get started.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
bootstrap()
|
|
print(f"[server] Listening on http://0.0.0.0:{PORT}")
|
|
app.run(host="0.0.0.0", port=PORT, threaded=True)
|