from __future__ import annotations

import logging
import socket
from pathlib import Path

from gi.repository import Gio, GLib

from .model import Connectivity, EthernetLink
from .notifier import ChangeNotifier

log = logging.getLogger("exo.network")

SYSFS_NET = Path("/sys/class/net")
NETLINK_ROUTE = 0
RTMGRP_LINK = 1
ARPHRD_ETHER = "1"

CONNECTIVITY_BY_GIO = {
    Gio.NetworkConnectivity.LOCAL: Connectivity.LOCAL_ONLY,
    Gio.NetworkConnectivity.LIMITED: Connectivity.LIMITED,
    Gio.NetworkConnectivity.PORTAL: Connectivity.PORTAL,
    Gio.NetworkConnectivity.FULL: Connectivity.FULL,
}


def _read(path: Path) -> str | None:
    try:
        return path.read_text().strip()
    except OSError:
        return None


def read_ethernet(root: Path = SYSFS_NET) -> tuple[EthernetLink, ...]:
    links = []
    for entry in sorted(root.iterdir()):
        is_physical_ethernet = (
            _read(entry / "type") == ARPHRD_ETHER
            and not (entry / "wireless").exists()
            and not (entry / "phy80211").exists()
            and "/virtual/" not in str(entry.resolve())
        )
        if not is_physical_ethernet:
            continue

        carrier = _read(entry / "carrier")
        connected = carrier == "1" if carrier is not None else _read(entry / "operstate") == "up"
        speed = _read(entry / "speed")
        links.append(
            EthernetLink(
                interface=entry.name,
                connected=connected,
                speed_mbps=int(speed) if connected and speed and int(speed) > 0 else None,
                mac=_read(entry / "address"),
            )
        )
    return tuple(links)


class LinkMonitor(ChangeNotifier):
    def __init__(self, sysfs: Path = SYSFS_NET, watch_netlink: bool = True):
        super().__init__()
        self._sysfs = sysfs
        self._netlink: socket.socket | None = None
        self.ethernet: tuple[EthernetLink, ...] = read_ethernet(sysfs)
        self.connectivity = Connectivity.UNKNOWN

        if watch_netlink:
            self._watch_netlink()
        self._watch_connectivity()

    def _watch_netlink(self):
        try:
            sock = socket.socket(
                socket.AF_NETLINK, socket.SOCK_RAW | socket.SOCK_NONBLOCK, NETLINK_ROUTE
            )
            sock.bind((0, RTMGRP_LINK))
        except OSError as error:
            log.warning("netlink unavailable, ethernet state will not update live: %s", error)
            return
        self._netlink = sock
        GLib.io_add_watch(sock.fileno(), GLib.PRIORITY_DEFAULT, GLib.IOCondition.IN, self._on_netlink)

    def _on_netlink(self, _fd, _condition):
        while True:
            try:
                self._netlink.recv(65536)
            except BlockingIOError:
                break
        self.refresh_ethernet()
        return GLib.SOURCE_CONTINUE

    def refresh_ethernet(self):
        current = read_ethernet(self._sysfs)
        if current != self.ethernet:
            self.ethernet = current
            self.notify_changed()

    def _watch_connectivity(self):
        monitor = Gio.NetworkMonitor.get_default()
        if monitor is None:
            return
        monitor.connect("notify::connectivity", lambda *_: self._on_connectivity(monitor))
        self._on_connectivity(monitor)

    def _on_connectivity(self, monitor: Gio.NetworkMonitor):
        current = CONNECTIVITY_BY_GIO.get(monitor.get_connectivity(), Connectivity.UNKNOWN)
        if current != self.connectivity:
            self.connectivity = current
            self.notify_changed()