mission-control-v2/backend/services/homelab/netzgeraete.py

"""Neue Geräte im Heimnetz (Ausbauplan „Später“, Punkt 21, seit 26.09.2026).

Geplant war AdGuard als Quelle. Nachgesehen am 26.09.: AdGuard sieht in diesem Netz fast nur die Fritzbox, die die
Anfragen der meisten Geräte weiterreicht. Der Ausführer sucht deshalb alle 30 Minuten das Netz der Proxmox-Brücke ab
(Ping an jede Adresse, dann die Nachbartabelle; nur lesend) und schickt die Geräte mit MAC, IP und Namen im Bericht mit
(host.netz). Hier wird jede MAC mit dem Register <Datenordner>/netz-bekannt.json verglichen:
  • Beim allerersten Abgleich gilt alles als bekannt — sonst stünden alle Geräte auf einmal da.
  • Eine neue MAC wird „neu“: Lucy meldet sie einmal (nicht dringend, nachts mit der Morgenmeldung), und der Wächter
    zeigt einen gelben Hinweis „Neues Gerät im Netz“ mit dem Knopf „Ist bekannt“, bis jemand ihn drückt.
  • Handys wechseln ihre MAC (zufällige, „lokal verwaltete“ Adressen). Trägt eine neue MAC denselben Namen wie ein
    bekanntes Gerät, gilt sie als dasselbe Gerät.
  • Die Gäste des Proxmox-Hosts kennt der Orchestrator ohnehin; ihre IPs gelten als bekannt.
Verschwundene Geräte bleiben im Register (mit „zuletzt gesehen“): Ein Gerät, das nur ab und zu da ist, soll nicht jedes
Mal neu sein. Geschrieben wird nur im Prozess der Oberfläche (Bericht, Knopf); der Wächter im Steward liest.
"""

import json
import logging
import threading
import time
from datetime import datetime
from pathlib import Path

from kern.einstellungen import einstellungen
from kern.zeit import LOCAL_TZ

from services import announce

log = logging.getLogger(__name__)

BETREFF = "[Homelab]"
_lock = threading.Lock()


def _pfad() -> Path:
    return einstellungen().daten_dir / "netz-bekannt.json"


def _lesen() -> dict | None:
    try:
        daten = json.loads(_pfad().read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None
    return daten if isinstance(daten, dict) and isinstance(daten.get("geraete"), dict) else None


def _schreiben(daten: dict) -> None:
    pfad = _pfad()
    pfad.parent.mkdir(parents=True, exist_ok=True)
    tmp = pfad.with_suffix(".tmp")
    tmp.write_text(json.dumps(daten, ensure_ascii=False, indent=1), encoding="utf-8")
    tmp.replace(pfad)


def lokal_verwaltet(mac: str) -> bool:
    """Zufällige MAC (zweites Bit des ersten Bytes gesetzt) — so schützen Handys ihre Privatsphäre."""
    try:
        return bool(int(mac.split(":")[0], 16) & 0x02)
    except (ValueError, IndexError):
        return False


def _kurzname(name: str | None) -> str | None:
    return name.removesuffix(".fritz.box") if name else None


def abgleichen(bericht: dict, jetzt: float | None = None) -> list[dict]:
    """Nach jedem Bericht: Register nachführen. Rückgabe: die Geräte, die eben neu dazukamen (für die Meldung)."""
    jetzt = time.time() if jetzt is None else jetzt
    b = bericht.get("bericht") if isinstance(bericht.get("bericht"), dict) else bericht
    netz = ((b or {}).get("host") or {}).get("netz")
    if not isinstance(netz, dict) or not isinstance(netz.get("geraete"), list) or not netz.get("netz"):
        return []
    gast_ips = {str(g.get("ip")) for g in (b or {}).get("gaeste") or [] if isinstance(g, dict) and g.get("ip")}
    neu: list[dict] = []
    with _lock:
        daten = _lesen()
        erstes_mal = daten is None
        daten = daten or {"geraete": {}}
        register: dict = daten["geraete"]
        namen = {e.get("name"): mac for mac, e in register.items() if e.get("name")}
        for g in netz["geraete"]:
            mac, ip, name = str(g.get("mac") or "").lower(), str(g.get("ip") or ""), g.get("name")
            if not mac or not ip:
                continue
            if mac in register:
                register[mac].update(ip=ip, zuletzt=jetzt, **({"name": name} if name else {}))
                continue
            gleicher_name = name and namen.get(name)
            bekannt = erstes_mal or ip in gast_ips or bool(gleicher_name)
            register[mac] = {"ip": ip, "name": name, "erstmals": jetzt, "zuletzt": jetzt, "bestaetigt": bekannt}
            if gleicher_name:
                register[mac]["wie"] = gleicher_name
            if not bekannt:
                neu.append({"mac": mac, **register[mac]})
        daten["netz"], daten["stand"] = netz["netz"], float(netz.get("zeit") or jetzt)
        _schreiben(daten)
    for g in neu:
        log.info("netz: neues Gerät %s (%s, %s)", g["ip"], g["mac"], g.get("name") or "ohne Namen")
    if neu:
        _melden(neu)
    return neu


def _beschreibung(g: dict) -> str:
    teile = [g["ip"], f"MAC {g['mac']}"]
    if lokal_verwaltet(g["mac"]):
        teile.append("zufällige MAC, oft ein Handy oder Tablet")
    return " · ".join(teile)


def _melden(neu: list[dict]) -> None:
    zeilen = [f"• {_kurzname(g.get('name')) or 'ohne Namen'} ({_beschreibung(g)})" for g in neu[:8]]
    text = (("Ein neues Gerät ist" if len(neu) == 1 else f"{len(neu)} neue Geräte sind") + " im Heimnetz aufgetaucht:\n"
            + "\n".join(zeilen) + "\nKennst du es, drück im Homelab-Cockpit „Ist bekannt“.")
    try:
        announce.notify_telegram(BETREFF, text)
    except Exception:
        log.warning("netz: Meldung ging nicht raus", exc_info=True)


def bestaetigen(mac: str) -> dict:
    """Knopf „Ist bekannt“ (Aktion des Hinweises)."""
    mac = str(mac or "").lower()
    with _lock:
        daten = _lesen()
        eintrag = (daten or {}).get("geraete", {}).get(mac)
        if not daten or not eintrag:
            return {"ok": False, "detail": "Dieses Gerät kennt das Register nicht."}
        eintrag["bestaetigt"] = True
        _schreiben(daten)
    return {"ok": True, "text": f"Gemerkt: {_kurzname(eintrag.get('name')) or eintrag['ip']} ist bekannt. Der Hinweis "
                                "verschwindet in wenigen Minuten."}


def lage() -> dict:
    """Für GET /api/homelab/netz: alle Geräte, neue zuerst, dann nach IP."""
    daten = _lesen() or {"geraete": {}}
    geraete = [{"mac": mac, **e, "name": _kurzname(e.get("name")), "zufaellig": lokal_verwaltet(mac)}
               for mac, e in daten["geraete"].items()]

    def ip_schluessel(g: dict) -> tuple:
        try:
            return tuple(int(t) for t in g["ip"].split("."))
        except (ValueError, AttributeError):
            return (999,)
    geraete.sort(key=lambda g: (g.get("bestaetigt", True), ip_schluessel(g)))
    return {"netz": daten.get("netz"), "stand": daten.get("stand"), "geraete": geraete}


def pruefe_netzgeraete() -> list:
    """Für den Wächter (Rolle homelab): jedes unbestätigte Gerät als gelber Hinweis mit dem Knopf „Ist bekannt“."""
    from services.waechter import Befund, _aktion

    befunde = []
    for mac, e in ((_lesen() or {}).get("geraete") or {}).items():
        if e.get("bestaetigt", True):
            continue
        g = {"mac": mac, **e}
        seit = datetime.fromtimestamp(float(e.get("erstmals") or 0), LOCAL_TZ).strftime("%d.%m. %H:%M")
        befunde.append(Befund(
            id=f"netz:{mac}", stufe="gelb", titel=f"Neues Gerät im Netz: {_kurzname(e.get('name')) or e.get('ip')}",
            text=f"{_beschreibung(g)}, zuerst gesehen {seit}. Kennst du es, drück „Ist bekannt“; sonst in der Fritzbox "
                 "nachsehen, wem es gehört.",
            quelle="netz", aktionen=[_aktion("bekannt", "Ist bekannt", mac=mac)], sofort=True))
    return befunde