# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""NAPALM driver for AVM FritzBox routers (TR-064 via fritzconnection).

FritzBox exposes its configuration and status through TR-064 (a UPnP/SOAP
based management protocol). Authentication uses the FritzBox device
username/password (System -> FRITZ!Box Users). Pass connection tuning via
*optional_args*:

    optional_args={
        "use_tls": True,   # use HTTPS (port 49443); False for HTTP (port 49000)
        "port": 49443,     # override the TR-064 port
    }

FritzBox always serves TR-064 over HTTPS with a self-signed certificate;
``fritzconnection`` does not verify it, so no separate ``verify`` option is
needed.

This driver is read-only: it does not implement NAPALM's configuration
management methods (``load_merge_candidate``, ``commit_config``, ...).
"""

from __future__ import annotations

import socket
from typing import Any

from fritzconnection import FritzConnection
from fritzconnection.core.exceptions import (
    FritzActionError,
    FritzArrayIndexError,
    FritzConnectionException,
    FritzServiceError,
)
from napalm.base.exceptions import ConnectionClosedException, ConnectionException

from napalm_device_types import FingerprintRule, ResidentialGatewayDriver
from napalm_device_types.models import (
    HostDict,
    NATTranslationDict,
    PortForwardDict,
    RadioStatusDict,
    SSIDDict,
    VPNTunnelDict,
    WANStatusDict,
    WirelessClientDict,
)

# Exceptions that indicate an optional/unavailable TR-064 service or action
# (e.g. a service not present on a given FritzOS version/model).
_OPTIONAL_SERVICE_ERRORS = (FritzServiceError, FritzActionError, FritzConnectionException)


class FritzBoxDriver(ResidentialGatewayDriver):
    """NAPALM driver for AVM FritzBox (read-only, TR-064)."""

    VENDOR = "AVM"
    DRIVER_NAME = "fritzbox"
    OUI_PREFIXES = [
        "00:04:0E",  # AVM GmbH — IEEE
        "3C:A6:2F",  # AVM Audiovisuelles Marketing — IEEE
        "9C:C7:A6",  # AVM GmbH — IEEE
        "B0:F2:08",  # AVM Audiovisuelles Marketing — IEEE
        "E0:28:6D",  # AVM Audiovisuelles Marketing — IEEE
    ]
    HTTP_FINGERPRINT = [
        FingerprintRule("fritz!box", weight=10.0, mandatory=True),
        FingerprintRule("fritzbox", weight=4.0),
        FingerprintRule("avm fritz", weight=4.0),
    ]

    _WAN_IP_SERVICES = ("WANIPConnection1", "WANPPPConnection1")
    _WLAN_SERVICES = ("WLANConfiguration1", "WLANConfiguration2", "WLANConfiguration3")

    def __init__(
        self,
        hostname: str,
        username: str,
        password: str,
        timeout: int = 60,
        optional_args: dict[str, Any] | None = None,
    ) -> None:
        self.hostname = hostname
        self.username = username
        self.password = password
        self.timeout = timeout
        self.optional_args = optional_args or {}

        # NAPALM standard attributes
        self.force_no_enable = True
        self.use_canonical_interface = False

        self.use_tls = bool(self.optional_args.get("use_tls", True))
        self.port = int(self.optional_args.get("port") or (49443 if self.use_tls else 49000))

        self.fc: FritzConnection | None = None

    # ------------------------------------------------------------------
    # Connection management
    # ------------------------------------------------------------------

    def open(self) -> None:
        """Open a TR-064 connection to the FritzBox and validate credentials."""
        try:
            fc = FritzConnection(
                address=self.hostname,
                port=self.port,
                user=self.username,
                password=self.password,
                use_tls=self.use_tls,
                timeout=self.timeout,
            )
            self.fc = fc
            # Lightweight connectivity and auth check
            self._call("DeviceInfo1", "GetInfo")
        except Exception as exc:
            self.fc = None
            raise ConnectionException(
                f"Cannot connect to FritzBox TR-064 at {self.hostname}:{self.port}: {exc}"
            ) from exc

    def close(self) -> None:
        """Drop the TR-064 connection."""
        self.fc = None

    def is_alive(self) -> dict[str, bool]:
        """Return whether the connection is usable.

        Performs a lightweight socket-level check without sending a full
        TR-064 request.
        """
        if self.fc is None:
            return {"is_alive": False}
        try:
            with socket.create_connection((self.hostname, self.port), timeout=5):
                pass
            return {"is_alive": True}
        except OSError:
            return {"is_alive": False}

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    def _call(self, service: str, action: str, **kwargs: Any) -> dict[str, Any]:
        """Invoke a TR-064 SOAP action and return its output arguments.

        :raises ConnectionClosedException: if called before :meth:`open`.
        """
        if self.fc is None:
            raise ConnectionClosedException("Not connected – call open() first.")
        return self.fc.call_action(service, action, **kwargs)

    def _wlan_services(self) -> list[str]:
        """Return the WLAN service names actually present on this device."""
        if self.fc is None:
            return []
        return [s for s in self._WLAN_SERVICES if s in self.fc.services]

    def _wan_ip_service(self) -> str:
        """Return whichever WAN IP connection service this device exposes."""
        if self.fc is not None:
            for service in self._WAN_IP_SERVICES:
                if service in self.fc.services:
                    return service
        raise ConnectionException("No WAN IP connection service found on this FritzBox")

    def _wan_external_ip(self) -> str:
        try:
            service = self._wan_ip_service()
            return self._call(service, "GetExternalIPAddress").get("NewExternalIPAddress", "")
        except _OPTIONAL_SERVICE_ERRORS:
            return ""

    def _wan_external_ipv6(self, service: str) -> str:
        try:
            return self._call(service, "X_AVM-DE_GetExternalIPv6Address").get(
                "NewExternalIPv6Address", ""
            )
        except _OPTIONAL_SERVICE_ERRORS:
            return ""

    # ------------------------------------------------------------------
    # NAPALM standard getters
    # ------------------------------------------------------------------

    def get_facts(self) -> dict[str, Any]:
        """Return a dictionary of general device facts.

        Calls ``DeviceInfo1.GetInfo``.
        """
        info = self._call("DeviceInfo1", "GetInfo")

        try:
            interface_list = list(self.get_interfaces().keys())
        except Exception:
            interface_list = []

        return {
            "vendor": self.VENDOR,
            "model": info.get("NewModelName", "") or self.VENDOR,
            "hostname": self.hostname,
            "fqdn": self.hostname,
            "os_version": info.get("NewSoftwareVersion", ""),
            "serial_number": info.get("NewSerialNumber", ""),
            "uptime": int(info.get("NewUpTime", 0)),
            "interface_list": interface_list,
        }

    def get_interfaces(self) -> dict[str, dict[str, Any]]:
        """Return interface details keyed by interface name.

        Combines ``LANEthernetInterfaceConfig1.GetInfo``,
        ``WANCommonInterfaceConfig1.GetCommonLinkProperties`` and
        ``WLANConfiguration{1,2,3}.GetInfo``.
        """
        interfaces: dict[str, dict[str, Any]] = {}

        try:
            lan = self._call("LANEthernetInterfaceConfig1", "GetInfo")
            interfaces["lan"] = {
                "is_up": lan.get("NewStatus", "") == "Up",
                "is_enabled": bool(lan.get("NewEnable", False)),
                "description": "LAN",
                "last_flapped": -1.0,
                "mac_address": (lan.get("NewMACAddress") or "").lower(),
                "speed": float(lan.get("NewMaxBitRate", 0) or 0),
                "mtu": 1500,
            }
        except _OPTIONAL_SERVICE_ERRORS:
            pass

        try:
            link = self._call("WANCommonInterfaceConfig1", "GetCommonLinkProperties")
            interfaces["wan"] = {
                "is_up": link.get("NewPhysicalLinkStatus", "") == "Up",
                "is_enabled": True,
                "description": link.get("NewWANAccessType", "WAN"),
                "last_flapped": -1.0,
                "mac_address": "",
                "speed": float(link.get("NewLayer1DownstreamMaxBitRate", 0) or 0) / 1000,
                "mtu": 1500,
            }
        except _OPTIONAL_SERVICE_ERRORS:
            pass

        for service in self._wlan_services():
            try:
                wlan = self._call(service, "GetInfo")
            except _OPTIONAL_SERVICE_ERRORS:
                continue
            interfaces[service.lower()] = {
                "is_up": wlan.get("NewStatus", "") == "Up",
                "is_enabled": bool(wlan.get("NewEnable", False)),
                "description": wlan.get("NewSSID", ""),
                "last_flapped": -1.0,
                "mac_address": (wlan.get("NewBSSID") or "").lower(),
                "speed": float(wlan.get("NewMaxBitRate", 0) or 0),
                "mtu": 1500,
            }

        return interfaces

    def get_interfaces_ip(self) -> dict[str, dict[str, Any]]:
        """Return IP addresses keyed by interface name.

        WAN address comes from ``GetExternalIPAddress`` (and, if available,
        ``X_AVM-DE_GetExternalIPv6Address``) on the active WAN connection
        service. The LAN address is the address used to reach the device
        (TR-064 does not expose the router's own LAN IP generically).
        """
        result: dict[str, dict[str, Any]] = {}

        try:
            service = self._wan_ip_service()
            ext_ip = self._call(service, "GetExternalIPAddress").get("NewExternalIPAddress", "")
            if ext_ip:
                result["wan"] = {"ipv4": {ext_ip: {"prefix_length": 32}}}
                ipv6 = self._wan_external_ipv6(service)
                if ipv6:
                    result["wan"]["ipv6"] = {ipv6: {"prefix_length": 64}}
        except _OPTIONAL_SERVICE_ERRORS:
            pass

        result["lan"] = {"ipv4": {self.hostname: {"prefix_length": 24}}}
        return result

    def get_arp_table(self, vrf: str = "") -> list[dict[str, Any]]:
        """Return the ARP table, derived from :meth:`get_hosts`.

        Each entry contains: ``interface``, ``mac``, ``ip``, ``age``.
        """
        arp_table: list[dict[str, Any]] = []
        for host in self.get_hosts():
            if not host["is_active"] or not host["mac"] or not host["ip"]:
                continue
            arp_table.append(
                {
                    "interface": host["interface_type"],
                    "mac": host["mac"],
                    "ip": host["ip"],
                    "age": 0.0,
                }
            )
        return arp_table

    # ------------------------------------------------------------------
    # Residential gateway extensions
    # ------------------------------------------------------------------

    def get_wan_status(self) -> WANStatusDict:
        """Return WAN/internet connection status.

        Calls ``WANCommonInterfaceConfig1.GetCommonLinkProperties`` and
        ``GetAddonInfos`` for line/traffic stats, plus ``GetStatusInfo`` and
        ``GetExternalIPAddress`` on the active WAN connection service.
        """
        link = self._call("WANCommonInterfaceConfig1", "GetCommonLinkProperties")

        try:
            addon = self._call("WANCommonInterfaceConfig1", "GetAddonInfos")
        except _OPTIONAL_SERVICE_ERRORS:
            addon = {}

        service = self._wan_ip_service()
        status = self._call(service, "GetStatusInfo")
        ext_ip = self._call(service, "GetExternalIPAddress").get("NewExternalIPAddress", "")

        result: WANStatusDict = {
            "connection_type": link.get("NewWANAccessType", ""),
            "is_connected": status.get("NewConnectionStatus", "") == "Connected",
            "external_ip": ext_ip,
            "uptime": int(status.get("NewUptime", 0) or 0),
            "bytes_sent": int(addon.get("NewTotalBytesSent", 0) or 0),
            "bytes_received": int(addon.get("NewTotalBytesReceived", 0) or 0),
            "max_bitrate_up": int(link.get("NewLayer1UpstreamMaxBitRate", 0) or 0) // 1000,
            "max_bitrate_down": int(link.get("NewLayer1DownstreamMaxBitRate", 0) or 0) // 1000,
            "link_status": link.get("NewPhysicalLinkStatus", ""),
        }

        ipv6 = self._wan_external_ipv6(service)
        if ipv6:
            result["external_ipv6"] = ipv6

        return result

    def get_port_forwards(self) -> list[PortForwardDict]:
        """Return configured port forwarding rules.

        Iterates ``GetGenericPortMappingEntry`` on the active WAN connection
        service until the device reports an out-of-range index.
        """
        service = self._wan_ip_service()
        forwards: list[PortForwardDict] = []
        index = 0
        while True:
            try:
                entry = self._call(service, "GetGenericPortMappingEntry", NewPortMappingIndex=index)
            except (FritzArrayIndexError, FritzConnectionException):
                break

            forward: PortForwardDict = {
                "name": entry.get("NewPortMappingDescription", ""),
                "protocol": entry.get("NewProtocol", ""),
                "external_port": int(entry.get("NewExternalPort", 0) or 0),
                "internal_ip": entry.get("NewInternalClient", ""),
                "internal_port": int(entry.get("NewInternalPort", 0) or 0),
                "enabled": bool(entry.get("NewPortMappingEnabled", False)),
            }
            remote_host = entry.get("NewRemoteHost", "")
            if remote_host:
                forward["remote_host"] = remote_host
            forwards.append(forward)
            index += 1

        return forwards

    def get_hosts(self) -> list[HostDict]:
        """Return hosts known to the FritzBox.

        Iterates ``Hosts1.GetGenericHostEntry`` for
        ``Hosts1.GetHostNumberOfEntries`` entries.
        """
        try:
            count = int(self._call("Hosts1", "GetHostNumberOfEntries").get(
                "NewHostNumberOfEntries", 0
            ) or 0)
        except _OPTIONAL_SERVICE_ERRORS:
            return []

        hosts: list[HostDict] = []
        for index in range(count):
            try:
                entry = self._call("Hosts1", "GetGenericHostEntry", NewIndex=index)
            except (FritzArrayIndexError, FritzConnectionException):
                continue

            host: HostDict = {
                "mac": (entry.get("NewMACAddress") or "").lower(),
                "ip": entry.get("NewIPAddress", ""),
                "hostname": entry.get("NewHostName", ""),
                "interface_type": entry.get("NewInterfaceType", ""),
                "is_active": bool(entry.get("NewActive", False)),
            }
            lease = entry.get("NewLeaseTimeRemaining")
            if lease is not None:
                host["lease_time_remaining"] = int(lease)
            hosts.append(host)

        return hosts

    def get_nat_translations(self) -> list[NATTranslationDict]:
        """Return NAT translation entries derived from active port forwards.

        TR-064 does not expose a live connection-tracking table; this
        derives static NAT-PT entries from the enabled port forwarding
        rules, which is the closest equivalent FritzBox provides.
        """
        ext_ip = self._wan_external_ip()
        translations: list[NATTranslationDict] = []
        for forward in self.get_port_forwards():
            if not forward["enabled"]:
                continue
            remote = forward.get("remote_host") or "0.0.0.0"
            translations.append(
                {
                    "protocol": forward["protocol"].lower(),
                    "inside_local": f"{forward['internal_ip']}:{forward['internal_port']}",
                    "inside_global": f"{ext_ip}:{forward['external_port']}",
                    "outside_local": f"{remote}:0",
                    "outside_global": f"{remote}:0",
                    "age": 0.0,
                }
            )
        return translations

    def get_vpn_tunnels(self) -> dict[str, VPNTunnelDict]:
        """Return VPN tunnel status.

        FritzOS does not expose VPN (IPsec/WireGuard) tunnel state through
        TR-064; returns an empty dict.
        """
        return {}

    def get_wireless_clients(self) -> list[WirelessClientDict]:
        """Return wireless clients associated with the built-in access point(s).

        Iterates ``GetGenericAssociatedDeviceInfo`` for each
        ``WLANConfiguration{1,2,3}`` service present on the device.
        """
        clients: list[WirelessClientDict] = []
        for service in self._wlan_services():
            try:
                ssid = self._call(service, "GetInfo").get("NewSSID", "")
                total = int(self._call(service, "GetTotalAssociations").get(
                    "NewTotalAssociations", 0
                ) or 0)
            except _OPTIONAL_SERVICE_ERRORS:
                continue

            for index in range(total):
                try:
                    dev = self._call(
                        service, "GetGenericAssociatedDeviceInfo", NewAssociatedDeviceIndex=index
                    )
                except (FritzArrayIndexError, FritzConnectionException):
                    continue

                client: WirelessClientDict = {
                    "mac": (dev.get("NewAssociatedDeviceMACAddress") or "").lower(),
                    "ssid": ssid,
                    "radio": service,
                    "signal": int(dev.get("NewX_AVM-DE_SignalStrength", 0) or 0),
                    "noise": 0,
                    "tx_rate": float(dev.get("NewX_AVM-DE_Speed", 0) or 0),
                    "rx_rate": 0.0,
                    "uptime": 0,
                }
                ip = dev.get("NewAssociatedDeviceIPAddress", "")
                if ip:
                    client["ip"] = ip
                clients.append(client)

        return clients

    def get_ssids(self) -> dict[str, SSIDDict]:
        """Return configured wireless networks (SSIDs).

        Calls ``GetInfo`` and ``GetTotalAssociations`` on each
        ``WLANConfiguration{1,2,3}`` service present on the device.
        """
        ssids: dict[str, SSIDDict] = {}
        for service in self._wlan_services():
            try:
                info = self._call(service, "GetInfo")
            except _OPTIONAL_SERVICE_ERRORS:
                continue

            ssid = info.get("NewSSID", "")
            if not ssid:
                continue

            try:
                clients = int(self._call(service, "GetTotalAssociations").get(
                    "NewTotalAssociations", 0
                ) or 0)
            except _OPTIONAL_SERVICE_ERRORS:
                clients = 0

            ssids[ssid] = {
                "enabled": bool(info.get("NewEnable", False)),
                "radio": service,
                "bssid": (info.get("NewBSSID") or "").lower(),
                "encryption": info.get("NewBeaconType", ""),
                "hidden": bool(info.get("NewX_AVM-DE_HiddenSSID", False)),
                "clients": clients,
            }

        return ssids

    def get_radio_status(self) -> dict[str, RadioStatusDict]:
        """Return wireless radio status.

        Calls ``GetInfo`` on each ``WLANConfiguration{1,2,3}`` service
        present on the device. ``channel_width``, ``tx_power`` and
        ``frequency`` are not exposed via TR-064 and are reported as ``0``.
        """
        radios: dict[str, RadioStatusDict] = {}
        for service in self._wlan_services():
            try:
                info = self._call(service, "GetInfo")
            except _OPTIONAL_SERVICE_ERRORS:
                continue

            channel = int(info.get("NewChannel", 0) or 0)
            radios[service] = {
                "enabled": bool(info.get("NewEnable", False)),
                "band": "2.4GHz" if channel <= 14 else "5GHz",
                "channel": channel,
                "channel_width": 0,
                "tx_power": 0,
                "frequency": 0.0,
            }

        return radios