photobooth/backend/wifi.py

"""Gestion WiFi via nmcli — scan, connexion, réseaux enregistrés."""

import json
import logging
import subprocess
from pathlib import Path

log = logging.getLogger("photobooth.wifi")

WIFI_PASSWORDS_FILE = Path(__file__).parent.parent / "data" / "wifi_passwords.json"


def _nmcli_split(line):
    """Split nmcli -t output, gère les ':' échappés par '\\'."""
    parts = []
    cur = []
    i = 0
    while i < len(line):
        if line[i] == '\\' and i + 1 < len(line):
            cur.append(line[i + 1])
            i += 2
        elif line[i] == ':':
            parts.append(''.join(cur))
            cur = []
            i += 1
        else:
            cur.append(line[i])
            i += 1
    parts.append(''.join(cur))
    return parts


def _charger_mdp() -> dict:
    if WIFI_PASSWORDS_FILE.exists():
        try:
            return json.loads(WIFI_PASSWORDS_FILE.read_text())
        except Exception:
            return {}
    return {}


def _sauvegarder_mdp(data: dict):
    WIFI_PASSWORDS_FILE.parent.mkdir(parents=True, exist_ok=True)
    WIFI_PASSWORDS_FILE.write_text(json.dumps(data, indent=2, ensure_ascii=False))


def wifi_status() -> dict:
    result = {"connecte": False, "ssid": None, "signal": None, "ip": None, "interface": None}
    try:
        r = subprocess.run(
            ["nmcli", "-t", "-f", "DEVICE,TYPE,STATE,CONNECTION", "device", "status"],
            capture_output=True, text=True, timeout=10,
        )
        for line in r.stdout.strip().split("\n"):
            p = _nmcli_split(line)
            if len(p) >= 4 and p[1] == "wifi" and p[2] == "connected":
                result.update(connecte=True, interface=p[0], ssid=p[3])
                break

        if result["connecte"] and result["interface"]:
            r2 = subprocess.run(
                ["nmcli", "-t", "-f", "IP4.ADDRESS", "device", "show", result["interface"]],
                capture_output=True, text=True, timeout=5,
            )
            for line in r2.stdout.strip().split("\n"):
                if "IP4.ADDRESS" in line:
                    result["ip"] = line.split(":", 1)[1].strip().split("/")[0]
                    break

            r3 = subprocess.run(
                ["nmcli", "-t", "-f", "ACTIVE,SIGNAL", "dev", "wifi", "list"],
                capture_output=True, text=True, timeout=5,
            )
            for line in r3.stdout.strip().split("\n"):
                p = _nmcli_split(line)
                if len(p) >= 2 and p[0] in ("yes", "oui") and p[1].isdigit():
                    result["signal"] = int(p[1])
                    break
    except Exception as e:
        log.warning(f"wifi_status: {e}")
    return result


def wifi_scan() -> list:
    saved_ssids = {n["ssid"] for n in wifi_saved_list()}
    networks = []
    try:
        r = subprocess.run(
            ["nmcli", "-t", "-f", "SSID,SIGNAL,SECURITY,ACTIVE", "dev", "wifi", "list", "--rescan", "yes"],
            capture_output=True, text=True, timeout=20,
        )
        merged = {}
        for line in r.stdout.strip().split("\n"):
            if not line.strip():
                continue
            p = _nmcli_split(line)
            if len(p) < 4:
                continue
            ssid = p[0]
            if not ssid:
                continue
            sig = int(p[1]) if p[1].isdigit() else 0
            is_active = p[3] in ("yes", "oui")
            if ssid in merged:
                if sig > merged[ssid]["signal"]:
                    merged[ssid]["signal"] = sig
                if is_active:
                    merged[ssid]["actif"] = True
            else:
                merged[ssid] = {
                    "ssid": ssid,
                    "signal": sig,
                    "securise": bool(p[2] and p[2] not in ("--", "")),
                    "securite": p[2] if p[2] not in ("--", "") else "",
                    "actif": is_active,
                    "enregistre": ssid in saved_ssids,
                }
        networks = list(merged.values())
        networks.sort(key=lambda n: (-n["actif"], -n["signal"]))
    except Exception as e:
        log.warning(f"wifi_scan: {e}")
    return networks


def wifi_connect(ssid: str, password: str | None = None) -> dict:
    mdp = _charger_mdp()

    if not password and ssid in mdp:
        password = mdp[ssid]

    # Réseau enregistré sans nouveau mot de passe → activer la connexion existante
    if not password:
        try:
            r = subprocess.run(
                ["nmcli", "connection", "up", ssid],
                capture_output=True, text=True, timeout=30,
            )
            if r.returncode == 0:
                return {"succes": True, "message": f"Connecté à {ssid}"}
        except Exception:
            pass

    if password:
        try:
            # Supprimer l'ancienne connexion pour éviter les doublons
            subprocess.run(["nmcli", "connection", "delete", ssid],
                           capture_output=True, text=True, timeout=5)
            r = subprocess.run(
                ["nmcli", "dev", "wifi", "connect", ssid, "password", password],
                capture_output=True, text=True, timeout=30,
            )
            if r.returncode == 0:
                mdp[ssid] = password
                _sauvegarder_mdp(mdp)
                return {"succes": True, "message": f"Connecté à {ssid}"}
            err = r.stderr.strip() or r.stdout.strip()
            if "secret" in err.lower() or "no suitable" in err.lower():
                return {"succes": False, "message": "Mot de passe incorrect"}
            return {"succes": False, "message": err}
        except subprocess.TimeoutExpired:
            return {"succes": False, "message": "Timeout — le réseau ne répond pas"}
        except Exception as e:
            return {"succes": False, "message": str(e)}

    return {"succes": False, "message": "Mot de passe requis"}


def wifi_saved_list() -> list:
    saved = []
    mdp = _charger_mdp()
    try:
        r = subprocess.run(
            ["nmcli", "-t", "-f", "NAME,TYPE", "connection", "show"],
            capture_output=True, text=True, timeout=5,
        )
        for line in r.stdout.strip().split("\n"):
            p = _nmcli_split(line)
            if len(p) >= 2 and p[1] == "802-11-wireless":
                saved.append({"ssid": p[0], "mdp_connu": p[0] in mdp})
    except Exception as e:
        log.warning(f"wifi_saved_list: {e}")
    return saved


def wifi_forget(ssid: str) -> dict:
    mdp = _charger_mdp()
    try:
        r = subprocess.run(
            ["nmcli", "connection", "delete", ssid],
            capture_output=True, text=True, timeout=10,
        )
        if ssid in mdp:
            del mdp[ssid]
            _sauvegarder_mdp(mdp)
        if r.returncode == 0:
            return {"succes": True}
        return {"succes": False, "message": r.stderr.strip()}
    except Exception as e:
        return {"succes": False, "message": str(e)}


def wifi_get_password(ssid: str) -> str | None:
    return _charger_mdp().get(ssid)


# --- Watchdog connectivité ---

_internet_ok = False


def check_internet(timeout: int = 5) -> bool:
    """Teste la connectivité internet (DNS + HTTP rapide)."""
    import socket
    try:
        socket.create_connection(("1.1.1.1", 53), timeout=timeout).close()
        return True
    except OSError:
        return False


def watchdog_tick() -> str | None:
    """Vérifie internet. Retourne 'restored' si passage offline→online, None sinon."""
    global _internet_ok
    now_ok = check_internet()
    if now_ok and not _internet_ok:
        _internet_ok = True
        log.info("Internet restauré — flush spool")
        return "restored"
    _internet_ok = now_ok
    if not now_ok:
        _tenter_reconnexion_wifi()
    return None


def _tenter_reconnexion_wifi():
    """Si déconnecté du WiFi, tente de se reconnecter à un réseau enregistré visible."""
    status = wifi_status()
    if status["connecte"]:
        return
    log.info("Pas de WiFi — scan des réseaux enregistrés")
    saved = {n["ssid"] for n in wifi_saved_list()}
    if not saved:
        return
    try:
        r = subprocess.run(
            ["nmcli", "-t", "-f", "SSID,SIGNAL", "dev", "wifi", "list", "--rescan", "yes"],
            capture_output=True, text=True, timeout=15,
        )
        candidates = []
        for line in r.stdout.strip().split("\n"):
            p = _nmcli_split(line)
            if len(p) >= 2 and p[0] in saved and p[1].isdigit():
                candidates.append((p[0], int(p[1])))
        candidates.sort(key=lambda x: -x[1])
        for ssid, sig in candidates:
            log.info(f"Tentative reconnexion WiFi: {ssid} (signal {sig}%)")
            result = wifi_connect(ssid)
            if result.get("succes"):
                log.info(f"Reconnecté à {ssid}")
                return
    except Exception as e:
        log.warning(f"Reconnexion WiFi échouée: {e}")