ARIA-AGENT/satellite/satellite.py

"""
ARIA Satellit — Info-/Gateway-Aussenposten in einem fremden Netz.

Laeuft eigenstaendig (z.B. im Buero) und verbindet sich als RVS-Client in
Stefans Raum (gleicher Token). Gibt ARIA damit Augen + Haende in DIESEM Netz:

  Augen: entdeckt Geraete (mDNS/Zeroconf, SSDP/UPnP + DIAL, ARP-Tabelle) und
         meldet ein Inventar → sat_devices.
  Haende: steuert Geraete (DIAL-App-Launch z.B. YouTube auf Fire TV, Wake-on-
          LAN, generisches HTTP) → sat_command / sat_result. Nur wenn
          CONTROL_ENABLED=true, Aktion in der Allowlist, alles geloggt.

Adressierung: mehrere Satelliten haengen im selben RVS-Raum. Jeder hat eine
SATELLITE_ID (technisch, eindeutig) + SATELLITE_LOCATION (menschlich, "Buero").
ARIA spricht einen Satelliten ueber seine ID/Location an.

Message-Typen (RVS, Base64/JSON-Relay wie der Rest):
  raus:  sat_hello   {id, location, caps, ts}
         sat_devices {requestId, satellite, devices:[...]}
         sat_result  {requestId, satellite, ok, result|error}
  rein:  sat_discover {satellite?, requestId}
         sat_command  {satellite?, requestId, device, action, params}

Sicherheit: reagiert nur auf den eigenen RVS-Raum (Token). Commands brauchen
CONTROL_ENABLED + Allowlist. Discovery ist read-only. Keine offenen Ports.
"""

from __future__ import annotations

import asyncio
import json
import logging
import os
import re
import socket
import struct
import time
from typing import Optional

import websockets

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [satellite] %(levelname)s %(message)s",
)
logger = logging.getLogger("satellite")


# ─── Konfiguration ──────────────────────────────────────────────────

def _env_bool(name: str, default: bool) -> bool:
    v = os.environ.get(name)
    if v is None:
        return default
    return v.strip().lower() in ("1", "true", "yes", "on", "ja")


def _default_id() -> str:
    host = socket.gethostname() or "satellite"
    slug = re.sub(r"[^a-zA-Z0-9_-]+", "-", host).strip("-").lower()
    return slug or "satellite"


RVS_HOST = os.environ.get("RVS_HOST", "")
RVS_PORT = int(os.environ.get("RVS_PORT", "443") or "443")
RVS_TLS = _env_bool("RVS_TLS", True)
RVS_TOKEN = os.environ.get("RVS_TOKEN", "")

SATELLITE_ID = (os.environ.get("SATELLITE_ID") or _default_id()).strip()
SATELLITE_LOCATION = (os.environ.get("SATELLITE_LOCATION") or SATELLITE_ID).strip()

CONTROL_ENABLED = _env_bool("CONTROL_ENABLED", False)
CONTROL_ALLOWLIST = [
    a.strip() for a in
    os.environ.get("CONTROL_ALLOWLIST", "dial.launch,wol,http.get").split(",")
    if a.strip()
]

SCAN_INTERVAL_SEC = int(os.environ.get("SCAN_INTERVAL_SEC", "300") or "300")
DISCOVER_TIMEOUT_SEC = float(os.environ.get("DISCOVER_TIMEOUT_SEC", "6") or "6")
DEVICE_CACHE_TTL_SEC = int(os.environ.get("DEVICE_CACHE_TTL_SEC", "120") or "120")

HEARTBEAT_SEC = 25

# mDNS-Servicetypen, die fuer ARIA interessant sind.
MDNS_TYPES = [
    "_googlecast._tcp.local.",      # Chromecast / Google TV / Nest
    "_airplay._tcp.local.",         # Apple TV / AirPlay
    "_raop._tcp.local.",            # AirPlay-Audio
    "_spotify-connect._tcp.local.",  # Spotify-Geraete
    "_sonos._tcp.local.",           # Sonos
    "_hap._tcp.local.",             # HomeKit
    "_printer._tcp.local.",         # Drucker
    "_ipp._tcp.local.",             # Drucker (IPP)
    "_smb._tcp.local.",             # NAS / Fileshares
    "_workstation._tcp.local.",     # generische Hosts
    "_http._tcp.local.",            # Web-UIs (Router, NAS, IoT)
]

CAPABILITIES = ["discover"]
if CONTROL_ENABLED:
    CAPABILITIES += CONTROL_ALLOWLIST


# ─── Discovery ──────────────────────────────────────────────────────

def _discover_mdns(timeout: float) -> list[dict]:
    """Blockierend (im Executor): mDNS/Zeroconf-Sweep ueber MDNS_TYPES."""
    out: dict[str, dict] = {}
    try:
        from zeroconf import Zeroconf, ServiceBrowser
    except Exception as exc:
        logger.warning("zeroconf nicht verfuegbar: %s", exc)
        return []

    class _Listener:
        def add_service(self, zc, type_, name):
            try:
                info = zc.get_service_info(type_, name, timeout=2000)
            except Exception:
                info = None
            if not info:
                return
            ips = []
            try:
                for addr in info.parsed_addresses():
                    ips.append(addr)
            except Exception:
                pass
            props = {}
            try:
                for k, v in (info.properties or {}).items():
                    try:
                        props[k.decode("utf-8", "ignore")] = (
                            v.decode("utf-8", "ignore") if isinstance(v, (bytes, bytearray)) else v)
                    except Exception:
                        pass
            except Exception:
                pass
            friendly = name.split("." + type_.split(".", 1)[0])[0].strip(".")
            fn = props.get("fn") or props.get("friendlyName") or friendly
            dev_id = _slug(f"{fn}-{ips[0] if ips else name}")
            out[dev_id] = {
                "id": dev_id,
                "name": fn,
                "type": _mdns_kind(type_),
                "ip": ips[0] if ips else "",
                "port": info.port,
                "via": "mdns",
                "service": type_,
                "model": props.get("md") or props.get("model") or "",
            }

        def update_service(self, *a):
            pass

        def remove_service(self, *a):
            pass

    zc = None
    try:
        zc = Zeroconf()
        listener = _Listener()
        for t in MDNS_TYPES:
            try:
                ServiceBrowser(zc, t, listener)
            except Exception:
                pass
        time.sleep(timeout)
    except Exception as exc:
        logger.warning("mDNS-Sweep-Fehler: %s", exc)
    finally:
        try:
            if zc:
                zc.close()
        except Exception:
            pass
    return list(out.values())


def _mdns_kind(service_type: str) -> str:
    m = {
        "_googlecast": "cast", "_airplay": "airplay", "_raop": "airplay-audio",
        "_spotify-connect": "spotify", "_sonos": "sonos", "_hap": "homekit",
        "_printer": "printer", "_ipp": "printer", "_smb": "fileshare",
        "_workstation": "host", "_http": "web",
    }
    for k, v in m.items():
        if service_type.startswith(k):
            return v
    return "unknown"


def _discover_ssdp(timeout: float) -> list[dict]:
    """Blockierend: SSDP M-SEARCH (UPnP + DIAL). Liefert v.a. Smart-TVs / Fire
    TV mit ihrer DIAL Application-URL (fuer App-Launch wie YouTube)."""
    out: dict[str, dict] = {}
    targets = [
        "urn:dial-multiscreen-org:service:dial:1",
        "ssdp:all",
    ]
    for st in targets:
        msg = (
            "M-SEARCH * HTTP/1.1\r\n"
            "HOST: 239.255.255.250:1900\r\n"
            'MAN: "ssdp:discover"\r\n'
            "MX: 2\r\n"
            f"ST: {st}\r\n\r\n"
        ).encode("utf-8")
        sock = None
        try:
            sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
            sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 2)
            sock.settimeout(timeout)
            sock.sendto(msg, ("239.255.255.250", 1900))
            deadline = time.time() + timeout
            while time.time() < deadline:
                try:
                    data, addr = sock.recvfrom(65507)
                except socket.timeout:
                    break
                except Exception:
                    break
                headers = _parse_http_headers(data.decode("utf-8", "ignore"))
                location = headers.get("location", "")
                dial_app = headers.get("application-url", "")
                ip = addr[0]
                dev = _fetch_upnp_description(location) if location else {}
                name = dev.get("name") or headers.get("server", "") or ip
                dev_id = _slug(f"{name}-{ip}")
                entry = out.get(dev_id, {
                    "id": dev_id, "name": name, "type": "media-renderer",
                    "ip": ip, "via": "ssdp",
                })
                if dev.get("name"):
                    entry["name"] = dev["name"]
                if dev.get("model"):
                    entry["model"] = dev["model"]
                if dev.get("manufacturer"):
                    entry["manufacturer"] = dev["manufacturer"]
                if dial_app or dev.get("dialAppUrl"):
                    entry["dialAppUrl"] = dial_app or dev.get("dialAppUrl")
                    entry["type"] = "dial"
                out[dev_id] = entry
        except Exception as exc:
            logger.debug("SSDP (%s) Fehler: %s", st, exc)
        finally:
            try:
                if sock:
                    sock.close()
            except Exception:
                pass
    return list(out.values())


def _fetch_upnp_description(location: str) -> dict:
    try:
        import requests
        r = requests.get(location, timeout=3)
        dial_app = r.headers.get("Application-URL", "")
        xml = r.text
        name = _xml_tag(xml, "friendlyName")
        model = _xml_tag(xml, "modelName")
        manuf = _xml_tag(xml, "manufacturer")
        return {"name": name, "model": model, "manufacturer": manuf, "dialAppUrl": dial_app}
    except Exception:
        return {}


def _discover_arp() -> list[dict]:
    """Rohe Host-Liste aus der ARP-Tabelle (kein aktiver Scan)."""
    out = []
    try:
        with open("/proc/net/arp", "r", encoding="utf-8") as f:
            lines = f.read().splitlines()[1:]
        for ln in lines:
            parts = ln.split()
            if len(parts) < 4:
                continue
            ip, _hw, _flags, mac = parts[0], parts[1], parts[2], parts[3]
            if mac == "00:00:00:00:00:00":
                continue
            out.append({
                "id": _slug(f"host-{ip}"), "name": ip, "type": "host",
                "ip": ip, "mac": mac, "via": "arp",
            })
    except Exception:
        pass
    return out


def _merge_devices(*lists) -> list[dict]:
    """Fuehrt Geraetelisten zusammen, dedupt per IP (reichere Quelle gewinnt)."""
    by_ip: dict[str, dict] = {}
    loose: list[dict] = []
    order = {"mdns": 3, "ssdp": 2, "arp": 1}
    for lst in lists:
        for d in lst:
            ip = d.get("ip") or ""
            if not ip:
                loose.append(d)
                continue
            cur = by_ip.get(ip)
            if not cur:
                by_ip[ip] = d
            else:
                # bessere Quelle / mehr Felder → mergen
                merged = {**d, **{k: v for k, v in cur.items() if v}}
                if order.get(d.get("via"), 0) >= order.get(cur.get("via"), 0):
                    merged.update({k: v for k, v in d.items() if v})
                # DIAL-URL / mac aus beiden behalten
                for key in ("dialAppUrl", "mac", "model", "manufacturer"):
                    merged[key] = d.get(key) or cur.get(key) or merged.get(key)
                by_ip[ip] = {k: v for k, v in merged.items() if v not in (None, "")}
    return list(by_ip.values()) + loose


# ─── Control ────────────────────────────────────────────────────────

async def _control(action: str, params: dict, devices: list[dict]) -> dict:
    """Fuehrt eine Steuer-Aktion aus. Guards: CONTROL_ENABLED + Allowlist."""
    if not CONTROL_ENABLED:
        return {"ok": False, "error": "Steuerung ist an diesem Satelliten deaktiviert (CONTROL_ENABLED=false)."}
    if action not in CONTROL_ALLOWLIST:
        return {"ok": False, "error": f"Aktion '{action}' nicht erlaubt (Allowlist: {', '.join(CONTROL_ALLOWLIST)})."}
    logger.info("[control] %s params=%s", action, {k: str(v)[:60] for k, v in (params or {}).items()})
    loop = asyncio.get_event_loop()
    try:
        if action == "dial.launch":
            return await loop.run_in_executor(None, _do_dial_launch, params, devices)
        if action == "wol":
            return await loop.run_in_executor(None, _do_wol, params)
        if action in ("http.get", "http.post"):
            return await loop.run_in_executor(None, _do_http, action, params)
        return {"ok": False, "error": f"Aktion '{action}' nicht implementiert."}
    except Exception as exc:
        return {"ok": False, "error": f"{action} fehlgeschlagen: {exc}"}


def _find_device(devices: list[dict], ref: str) -> Optional[dict]:
    ref = (ref or "").strip().lower()
    if not ref:
        return None
    for d in devices:
        if d.get("id", "").lower() == ref or d.get("ip", "") == ref:
            return d
    for d in devices:
        if ref in (d.get("name", "").lower()):
            return d
    return None


def _do_dial_launch(params: dict, devices: list[dict]) -> dict:
    """DIAL-App-Launch, z.B. YouTube-Video auf Fire TV / Smart-TV.
    params: {device, app='YouTube', v=<videoId> (oder beliebige app-params)}"""
    import requests
    ref = params.get("device") or ""
    dev = _find_device(devices, ref)
    app_url = (dev or {}).get("dialAppUrl") if dev else params.get("dialAppUrl")
    if not app_url:
        return {"ok": False, "error": f"Kein DIAL-Geraet fuer '{ref}' gefunden (oder keine Application-URL)."}
    app = params.get("app") or "YouTube"
    # app-Parameter (alles ausser device/app) als form-urlencoded Body.
    body = {k: v for k, v in (params or {}).items() if k not in ("device", "app", "dialAppUrl")}
    url = app_url.rstrip("/") + "/" + app
    r = requests.post(url, data=body, timeout=5)
    ok = r.status_code in (200, 201)
    return {"ok": ok, "result": f"DIAL {app} → {(dev or {}).get('name', ref)} (HTTP {r.status_code})"
            if ok else None,
            "error": None if ok else f"DIAL-Launch HTTP {r.status_code}: {r.text[:120]}"}


def _do_wol(params: dict) -> dict:
    mac = (params.get("mac") or "").strip()
    if not re.match(r"^([0-9A-Fa-f]{2}[:-]){5}[0-9A-Fa-f]{2}$", mac):
        return {"ok": False, "error": f"Ungueltige MAC: {mac!r}"}
    clean = re.sub(r"[:-]", "", mac)
    packet = b"\xff" * 6 + bytes.fromhex(clean) * 16
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
    s.sendto(packet, ("255.255.255.255", 9))
    s.close()
    return {"ok": True, "result": f"Wake-on-LAN an {mac} gesendet."}


def _do_http(action: str, params: dict) -> dict:
    import requests
    url = params.get("url") or ""
    if not url.startswith(("http://", "https://")):
        return {"ok": False, "error": "url (http/https) erforderlich."}
    method = "GET" if action == "http.get" else "POST"
    r = requests.request(method, url, data=params.get("body"),
                         headers=params.get("headers"), timeout=6)
    return {"ok": True, "result": {"status": r.status_code, "body": r.text[:2000]}}


# ─── Helpers ────────────────────────────────────────────────────────

def _slug(s: str) -> str:
    s = (s or "").strip().lower()
    s = re.sub(r"[^a-z0-9]+", "-", s).strip("-")
    return s or "dev"


def _parse_http_headers(text: str) -> dict:
    headers = {}
    for line in text.split("\r\n")[1:]:
        if ":" in line:
            k, _, v = line.partition(":")
            headers[k.strip().lower()] = v.strip()
    return headers


def _xml_tag(xml: str, tag: str) -> str:
    m = re.search(rf"<{tag}>(.*?)</{tag}>", xml, re.IGNORECASE | re.DOTALL)
    return m.group(1).strip() if m else ""


# ─── Satellit (RVS-Client) ──────────────────────────────────────────

class Satellite:
    def __init__(self) -> None:
        self.ws: Optional[websockets.WebSocketClientProtocol] = None
        self._devices: list[dict] = []
        self._devices_ts: float = 0.0
        self._scanning = False

    async def _scan(self, force: bool = False) -> list[dict]:
        fresh = (time.time() - self._devices_ts) < DEVICE_CACHE_TTL_SEC
        if self._devices and fresh and not force:
            return self._devices
        if self._scanning:
            # Laufenden Scan abwarten (grob)
            for _ in range(30):
                await asyncio.sleep(0.2)
                if not self._scanning:
                    break
            return self._devices
        self._scanning = True
        try:
            loop = asyncio.get_event_loop()
            mdns = await loop.run_in_executor(None, _discover_mdns, DISCOVER_TIMEOUT_SEC)
            ssdp = await loop.run_in_executor(None, _discover_ssdp, DISCOVER_TIMEOUT_SEC)
            arp = await loop.run_in_executor(None, _discover_arp)
            self._devices = _merge_devices(mdns, ssdp, arp)
            self._devices_ts = time.time()
            logger.info("[scan] %d Geraete (mdns=%d ssdp=%d arp=%d)",
                        len(self._devices), len(mdns), len(ssdp), len(arp))
        finally:
            self._scanning = False
        return self._devices

    async def _send(self, message: dict) -> None:
        if self.ws is None:
            return
        try:
            await self.ws.send(json.dumps(message))
        except Exception as exc:
            logger.warning("send fehlgeschlagen: %s", exc)

    async def _hello(self, log: bool = False) -> None:
        await self._send({
            "type": "sat_hello",
            "payload": {
                "id": SATELLITE_ID,
                "location": SATELLITE_LOCATION,
                "caps": CAPABILITIES,
                "control": CONTROL_ENABLED,
            },
            "timestamp": int(time.time() * 1000),
        })
        if log:
            logger.info("sat_hello gesendet: id=%s location=%s caps=%s",
                        SATELLITE_ID, SATELLITE_LOCATION, CAPABILITIES)

    def _for_me(self, payload: dict) -> bool:
        target = (payload.get("satellite") or "").strip().lower()
        if not target or target in ("all", "*"):
            return True
        return target in (SATELLITE_ID.lower(), SATELLITE_LOCATION.lower())

    async def _handle(self, raw: str) -> None:
        try:
            msg = json.loads(raw)
        except Exception:
            return
        mtype = msg.get("type", "")
        payload = msg.get("payload", {}) or {}

        if mtype == "sat_discover":
            if not self._for_me(payload):
                return
            req_id = payload.get("requestId", "")
            devices = await self._scan(force=bool(payload.get("force")))
            await self._send({
                "type": "sat_devices",
                "payload": {"requestId": req_id, "satellite": SATELLITE_ID,
                            "location": SATELLITE_LOCATION, "devices": devices},
                "timestamp": int(time.time() * 1000),
            })

        elif mtype == "sat_command":
            if not self._for_me(payload):
                return
            req_id = payload.get("requestId", "")
            action = (payload.get("action") or "").strip()
            params = payload.get("params") or {}
            if payload.get("device") and "device" not in params:
                params["device"] = payload.get("device")
            devices = await self._scan()
            result = await _control(action, params, devices)
            await self._send({
                "type": "sat_result",
                "payload": {"requestId": req_id, "satellite": SATELLITE_ID, **result},
                "timestamp": int(time.time() * 1000),
            })

    async def _periodic_scan(self) -> None:
        while True:
            try:
                await self._scan(force=True)
            except Exception as exc:
                logger.warning("periodischer Scan-Fehler: %s", exc)
            await asyncio.sleep(SCAN_INTERVAL_SEC)

    async def _heartbeat(self) -> None:
        # Re-announce bei jedem Heartbeat: falls die Bridge NACH uns (neu)
        # verbindet, lernt sie uns so innerhalb von HEARTBEAT_SEC — RVS replayt
        # nichts. Haelt zugleich last_seen in der Bridge-Registry frisch.
        while True:
            await asyncio.sleep(HEARTBEAT_SEC)
            await self._send({"type": "heartbeat", "timestamp": int(time.time() * 1000)})
            await self._hello()

    async def run(self) -> None:
        if not RVS_HOST or not RVS_TOKEN:
            logger.error("RVS_HOST und RVS_TOKEN sind Pflicht (siehe .env.example).")
            return
        asyncio.create_task(self._periodic_scan())
        backoff = 1
        while True:
            proto = "wss" if RVS_TLS else "ws"
            url = f"{proto}://{RVS_HOST}:{RVS_PORT}?token={RVS_TOKEN}"
            try:
                logger.info("Verbinde mit RVS %s://%s:%s …", proto, RVS_HOST, RVS_PORT)
                async with websockets.connect(url, max_size=8 * 1024 * 1024,
                                              ping_interval=20, ping_timeout=20) as ws:
                    self.ws = ws
                    backoff = 1
                    await self._hello(log=True)
                    hb = asyncio.create_task(self._heartbeat())
                    try:
                        async for raw in ws:
                            await self._handle(raw)
                    finally:
                        hb.cancel()
            except Exception as exc:
                logger.warning("RVS-Verbindung verloren: %s", exc)
            finally:
                self.ws = None
            await asyncio.sleep(backoff)
            backoff = min(backoff * 2, 30)


def main() -> None:
    logger.info("ARIA Satellit startet — id=%s location=%s control=%s",
                SATELLITE_ID, SATELLITE_LOCATION, CONTROL_ENABLED)
    try:
        asyncio.run(Satellite().run())
    except KeyboardInterrupt:
        pass


if __name__ == "__main__":
    main()