test_dashboard/server/app.py

import os
import threading
from datetime import datetime, timedelta, timezone
from functools import wraps
from pathlib import Path

import jwt
from dotenv import load_dotenv
from flask import Flask, g, jsonify, request, send_from_directory
from flask_cors import CORS
from werkzeug.security import check_password_hash, generate_password_hash

from db_py import (
    count_tests,
    count_users,
    create_user,
    del_config,
    get_all_tests,
    get_all_users,
    get_config,
    get_user_by_id,
    get_user_by_username,
    set_config,
    clear_users,
)
from scanner import full_scan, is_scan_in_progress, resolve_runtime_path, scan_results_only

BASE_DIR = Path(__file__).resolve().parent
load_dotenv(BASE_DIR / ".env")

PORT = int(os.getenv("PORT", "3001"))
JWT_SECRET = os.getenv("JWT_SECRET")
JWT_ALGORITHM = os.getenv("JWT_ALGORITHM", "HS256")
JWT_EXPIRES_HOURS = int(os.getenv("JWT_EXPIRES_HOURS", "8"))
DEFAULT_ADMIN_USERNAME = os.getenv("DEFAULT_ADMIN_USERNAME")
DEFAULT_ADMIN_PASSWORD = os.getenv("DEFAULT_ADMIN_PASSWORD")
DEFAULT_VIEWER_USERNAME = os.getenv("DEFAULT_VIEWER_USERNAME")
DEFAULT_VIEWER_PASSWORD = os.getenv("DEFAULT_VIEWER_PASSWORD")

ALLOWED_KEYS = {
    "target_dir",
    "results_dir",
    "results_dir_ref",
    "avg_time_coe",
    "avg_time_p2p",
    "avg_time_p3p",
    "scan_exclusions",
    "smb_username",
    "smb_password",
    "smb_domain",
}

DIST_DIR = BASE_DIR.parent / "dashboard" / "dist"

app = Flask(__name__, static_folder=str(DIST_DIR), static_url_path="")
CORS(app)


def _make_token(user):
    now = datetime.now(timezone.utc)
    payload = {
        "sub": str(user["id"]),
        "username": user["username"],
        "role": user["role"],
        "iat": int(now.timestamp()),
        "exp": int((now + timedelta(hours=JWT_EXPIRES_HOURS)).timestamp()),
    }
    return jwt.encode(payload, JWT_SECRET, algorithm=JWT_ALGORITHM)


def _decode_token(token):
    try:
        return jwt.decode(token, JWT_SECRET, algorithms=[JWT_ALGORITHM])
    except jwt.InvalidTokenError:
        return None


def _extract_bearer_token():
    auth_header = request.headers.get("Authorization", "")
    if not auth_header.lower().startswith("bearer "):
        return None
    return auth_header[7:].strip() or None


def require_auth(fn):
    @wraps(fn)
    def wrapper(*args, **kwargs):
        token = _extract_bearer_token()
        if not token:
            return jsonify({"error": "Authentication required"}), 401

        payload = _decode_token(token)
        if payload is None:
            return jsonify({"error": "Invalid or expired token"}), 401

        try:
            user_id = int(payload.get("sub"))
        except (TypeError, ValueError):
            return jsonify({"error": "Invalid token subject"}), 401

        user = get_user_by_id(user_id)
        if not user or not user.get("is_active"):
            return jsonify({"error": "User is not authorized"}), 401

        g.current_user = {
            "id": user["id"],
            "username": user["username"],
            "role": user["role"],
        }
        return fn(*args, **kwargs)

    return wrapper


def require_role(required_role):
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            current_user = getattr(g, "current_user", None)
            if not current_user:
                return jsonify({"error": "Authentication required"}), 401
            if current_user.get("role") != required_role:
                return jsonify({"error": "Forbidden"}), 403
            return fn(*args, **kwargs)

        return wrapper

    return decorator


def _create_default_user_if_missing(username, password, role):
    if not username or not password:
        print(f"[server] Skipping default {role} seed: username/password not configured")
        return

    existing = get_user_by_username(username)
    if existing:
        return

    create_user(
        username,
        generate_password_hash(password),
        role=role,
        is_active=1,
    )
    print(f"[server] Created default {role} user: {username}")


def _ensure_default_users():
    if count_users() == 0:
        print("[server] No users found. Seeding default accounts...")

    _create_default_user_if_missing(DEFAULT_ADMIN_USERNAME, DEFAULT_ADMIN_PASSWORD, "admin")
    _create_default_user_if_missing(DEFAULT_VIEWER_USERNAME, DEFAULT_VIEWER_PASSWORD, "viewer")


def _apply_smb_env_from_config():
    mapping = {
        "SMB_USERNAME": get_config("smb_username"),
        "SMB_PASSWORD": get_config("smb_password"),
        "SMB_DOMAIN": get_config("smb_domain"),
    }

    for env_key, value in mapping.items():
        if value in (None, ""):
            os.environ.pop(env_key, None)
        else:
            os.environ[env_key] = str(value)


@app.get("/api/tests")
@require_auth
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")
@require_auth
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/scan-status")
@require_auth
def get_scan_status_route():
    return jsonify({"scanning": is_scan_in_progress()})


@app.post("/api/auth/login")
def auth_login_route():
    body = request.get_json(silent=True)
    if not isinstance(body, dict):
        return jsonify({"error": "Request body must be a JSON object"}), 400

    username = (body.get("username") or "").strip()
    password = body.get("password") or ""

    if not username or not password:
        return jsonify({"error": "Username and password are required"}), 400

    user = get_user_by_username(username)
    if not user or not user.get("is_active"):
        return jsonify({"error": "Invalid username or password"}), 401

    if not check_password_hash(user["password_hash"], password):
        return jsonify({"error": "Invalid username or password"}), 401

    token = _make_token(user)
    return jsonify(
        {
            "token": token,
            "user": {
                "id": user["id"],
                "username": user["username"],
                "role": user["role"],
            },
        }
    )


@app.get("/api/auth/me")
@require_auth
def auth_me_route():
    return jsonify({"user": g.current_user})


@app.get("/api/users")
@require_auth
@require_role("admin")
def list_users_route():
    return jsonify(get_all_users())


@app.post("/api/users")
@require_auth
@require_role("admin")
def create_user_route():
    body = request.get_json(silent=True)
    if not isinstance(body, dict):
        return jsonify({"error": "Request body must be a JSON object"}), 400

    username = (body.get("username") or "").strip()
    password = body.get("password") or ""
    role = (body.get("role") or "viewer").strip().lower()

    if not username or not password:
        return jsonify({"error": "Username and password are required"}), 400

    if role not in {"admin", "viewer"}:
        return jsonify({"error": "Role must be admin or viewer"}), 400

    if get_user_by_username(username):
        return jsonify({"error": "User already exists"}), 409

    user_id = create_user(username, generate_password_hash(password), role=role, is_active=1)
    return jsonify({"id": user_id, "username": username, "role": role, "is_active": 1}), 201


@app.get("/api/config")
@require_auth
@require_role("admin")
def get_config_route():
    config = {}
    for key in ALLOWED_KEYS:
        config[key] = get_config(key)
    return jsonify(config)


@app.post("/api/config")
@require_auth
@require_role("admin")
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

    rescan_required = 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", "results_dir_ref", "scan_exclusions"}:
            rescan_required = True

    if rescan_required:
        _apply_smb_env_from_config()
        target_dir = resolve_runtime_path(get_config("target_dir"))
        results_dir = resolve_runtime_path(get_config("results_dir"))
        results_dir_ref = resolve_runtime_path(get_config("results_dir_ref"))
        if not target_dir or not results_dir:
            return jsonify({"error": "Directories not configured"}), 400

        if is_scan_in_progress():
            return jsonify({"ok": True, "scanning": True, "testCount": None, "completedCount": None})

        threading.Thread(
            target=full_scan,
            args=(target_dir, results_dir, results_dir_ref),
            daemon=True,
        ).start()
        return jsonify({"ok": True, "scanning": True, "testCount": None, "completedCount": None})

    return jsonify({"ok": True, "testCount": None, "completedCount": None})


@app.post("/api/config/rescan")
@require_auth
@require_role("admin")
def rescan_route():
    _apply_smb_env_from_config()
    target_dir = resolve_runtime_path(get_config("target_dir"))
    results_dir = resolve_runtime_path(get_config("results_dir"))
    results_dir_ref = resolve_runtime_path(get_config("results_dir_ref"))
    if not target_dir or not results_dir:
        return jsonify({"error": "Directories not configured"}), 400

    if is_scan_in_progress():
        return jsonify({"ok": True, "scanning": True, "testCount": None, "completedCount": None})

    threading.Thread(
        target=full_scan,
        args=(target_dir, results_dir, results_dir_ref),
        daemon=True,
    ).start()
    return jsonify({"ok": True, "scanning": True, "testCount": None, "completedCount": None})


@app.post("/api/config/rescan-results")
@require_auth
@require_role("admin")
def rescan_results_route():
    _apply_smb_env_from_config()
    results_dir = resolve_runtime_path(get_config("results_dir"))
    results_dir_ref = resolve_runtime_path(get_config("results_dir_ref"))
    if not results_dir:
        return jsonify({"error": "Results directory not configured"}), 400

    if is_scan_in_progress():
        return jsonify({"ok": True, "scanning": True, "testCount": None, "completedCount": None})

    threading.Thread(
        target=scan_results_only,
        args=(results_dir, results_dir_ref),
        daemon=True,
    ).start()
    return jsonify({"ok": True, "scanning": True, "testCount": None, "completedCount": None})


@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():
    _ensure_default_users()
    _apply_smb_env_from_config()
    target_dir = resolve_runtime_path(get_config("target_dir"))
    results_dir = resolve_runtime_path(get_config("results_dir"))
    results_dir_ref = resolve_runtime_path(get_config("results_dir_ref"))

    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, results_dir_ref)
            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")
    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)