Android-pyApi/android_pyapi/android/net_ext.py

"""
Android net 网络扩展组件 — Network, NetworkCapabilities, NetworkRequest,
NetworkAgent, LinkProperties, TrafficStats, ProxyInfo, DhcpInfo,
ConnectivityDiagnosticsManager, ConnectivityManager 扩展等
方法签名完全对齐 Android SDK
"""

from __future__ import annotations

import os
import socket
import struct
import time
from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union

from .._core import Stub, AndroidOnlyError

# Linux 真实实现导入
try:
    from .._linux_impl import IS_LINUX, LinuxNetwork, HAS_IP
    _USE_LINUX = IS_LINUX
except ImportError:
    _USE_LINUX = False

__all__ = [
    "Network", "NetworkCapabilities", "NetworkRequest", "NetworkSpecifier",
    "MatchAllNetworkSpecifier", "NetworkAgent", "NetworkInfo",
    "LinkProperties", "LinkAddress", "LinkQualityInfo",
    "TrafficStats", "ProxyInfo", "DhcpInfo", "DhcpInfoInternal",
    "WifiLinkQualityInfo", "EthernetLinkQualityInfo", "MobileLinkQualityInfo",
    "NetworkBadging", "ConnectivityDiagnosticsManager",
    "ConnectivityManager",
]


# ═══════════════════════════════════════════════════════════════
# Network
# ═══════════════════════════════════════════════════════════════

class Network(Stub):
    """Network — 网络"""

    def __init__(self, netId: int = 0):
        super().__init__()
        self._net_id = netId
        self._capabilities: Optional["NetworkCapabilities"] = None
        self._link_properties: Optional["LinkProperties"] = None

    def getNetId(self) -> int:
        return self._net_id

    def getNetworkCapabilities(self) -> Optional["NetworkCapabilities"]:
        return self._capabilities

    def getLinkProperties(self) -> Optional["LinkProperties"]:
        return self._link_properties

    def setCapabilities(self, caps: "NetworkCapabilities"):
        self._capabilities = caps

    def setLinkProperties(self, props: "LinkProperties"):
        self._link_properties = props

    def bindSocket(self, socketObj: Any):
        """绑定 socket 到该网络"""
        raise AndroidOnlyError("Network.bindSocket() is not available on desktop")

    def toString(self) -> str:
        return f"Network(netId={self._net_id})"

    def equals(self, other: Any) -> bool:
        if not isinstance(other, Network):
            return False
        return self._net_id == other._net_id

    def hashCode(self) -> int:
        return self._net_id & 0x7FFFFFFF


# ═══════════════════════════════════════════════════════════════
# NetworkCapabilities
# ═══════════════════════════════════════════════════════════════

class NetworkCapabilities(Stub):
    """NetworkCapabilities — 网络能力"""

    TRANSPORT_CELLULAR = 0
    TRANSPORT_WIFI = 1
    TRANSPORT_BLUETOOTH = 2
    TRANSPORT_ETHERNET = 3
    TRANSPORT_VPN = 4
    TRANSPORT_WIFI_AWARE = 5
    TRANSPORT_LOWPAN = 6
    TRANSPORT_TEST = 7
    TRANSPORT_USB = 8

    NET_CAPABILITY_MMS = 0
    NET_CAPABILITY_SUPL = 1
    NET_CAPABILITY_DUN = 2
    NET_CAPABILITY_FOTA = 3
    NET_CAPABILITY_IMS = 4
    NET_CAPABILITY_CBS = 5
    NET_CAPABILITY_WIFI_P2P = 6
    NET_CAPABILITY_IA = 7
    NET_CAPABILITY_RCS = 8
    NET_CAPABILITY_XCAP = 9
    NET_CAPABILITY_EIMS = 10
    NET_CAPABILITY_NOT_METERED = 11
    NET_CAPABILITY_INTERNET = 12
    NET_CAPABILITY_NOT_RESTRICTED = 13
    NET_CAPABILITY_TRUSTED = 14
    NET_CAPABILITY_NOT_VPN = 15
    NET_CAPABILITY_VALIDATED = 16
    NET_CAPABILITY_CAPTIVE_PORTAL = 17
    NET_CAPABILITY_NOT_ROAMING = 18
    NET_CAPABILITY_FOREGROUND = 19
    NET_CAPABILITY_NOT_CONGESTED = 20
    NET_CAPABILITY_NOT_SUSPENDED = 21
    NET_CAPABILITY_OEM_PAID = 22
    NET_CAPABILITY_MCX = 23
    NET_CAPABILITY_PARTIAL_CONNECTIVITY = 24
    NET_CAPABILITY_TEMPORARILY_NOT_METERED = 25
    NET_CAPABILITY_HEAD_UNIT = 26
    NET_CAPABILITY_NOT_CARRIER_PRIVILEGED = 27
    NET_CAPABILITY_BIP = 28

    SIGNAL_STRENGTH_UNSPECIFIED = -2147483648

    def __init__(self):
        super().__init__()
        self._transport_types: Set[int] = set()
        self._capabilities: Set[int] = set()
        self._link_down_bandwidth_kbps = 0
        self._link_up_bandwidth_kbps = 0
        self._signal_strength = NetworkCapabilities.SIGNAL_STRENGTH_UNSPECIFIED
        self._network_specifier: Optional["NetworkSpecifier"] = None
        self._owner_uid = -1
        self._ssid: Optional[str] = None

    def addTransportType(self, transportType: int) -> "NetworkCapabilities":
        self._transport_types.add(transportType)
        return self

    def removeTransportType(self, transportType: int) -> "NetworkCapabilities":
        self._transport_types.discard(transportType)
        return self

    def hasTransport(self, transportType: int) -> bool:
        return transportType in self._transport_types

    def addCapability(self, capability: int) -> "NetworkCapabilities":
        self._capabilities.add(capability)
        return self

    def removeCapability(self, capability: int) -> "NetworkCapabilities":
        self._capabilities.discard(capability)
        return self

    def hasCapability(self, capability: int) -> bool:
        return capability in self._capabilities

    def hasAllCapabilities(self, capabilities: List[int]) -> bool:
        return all(c in self._capabilities for c in capabilities)

    def getTransportTypes(self) -> List[int]:
        return sorted(self._transport_types)

    def getCapabilities(self) -> List[int]:
        return sorted(self._capabilities)

    def setLinkUpstreamBandwidthKbps(self, up: int) -> "NetworkCapabilities":
        self._link_up_bandwidth_kbps = up
        return self

    def setLinkDownstreamBandwidthKbps(self, down: int) -> "NetworkCapabilities":
        self._link_down_bandwidth_kbps = down
        return self

    def getLinkUpstreamBandwidthKbps(self) -> int:
        return self._link_up_bandwidth_kbps

    def getLinkDownstreamBandwidthKbps(self) -> int:
        return self._link_down_bandwidth_kbps

    def setSignalStrength(self, signalStrength: int) -> "NetworkCapabilities":
        self._signal_strength = signalStrength
        return self

    def getSignalStrength(self) -> int:
        return self._signal_strength

    def setNetworkSpecifier(self, specifier: "NetworkSpecifier") -> "NetworkCapabilities":
        self._network_specifier = specifier
        return self

    def getNetworkSpecifier(self) -> Optional["NetworkSpecifier"]:
        return self._network_specifier

    def setOwnerUid(self, uid: int) -> "NetworkCapabilities":
        self._owner_uid = uid
        return self

    def getOwnerUid(self) -> int:
        return self._owner_uid

    def setSsid(self, ssid: str) -> "NetworkCapabilities":
        self._ssid = ssid
        return self

    def getSsid(self) -> Optional[str]:
        return self._ssid

    def describeCapabilities(self) -> str:
        names = {
            0: "MMS", 1: "SUPL", 2: "DUN", 3: "FOTA", 4: "IMS",
            5: "CBS", 6: "WIFI_P2P", 7: "IA", 8: "RCS", 9: "XCAP",
            10: "EIMS", 11: "NOT_METERED", 12: "INTERNET",
            13: "NOT_RESTRICTED", 14: "TRUSTED", 15: "NOT_VPN",
            16: "VALIDATED", 17: "CAPTIVE_PORTAL", 18: "NOT_ROAMING",
            19: "FOREGROUND", 20: "NOT_CONGESTED", 21: "NOT_SUSPENDED",
        }
        return ", ".join(names.get(c, str(c)) for c in sorted(self._capabilities))

    def toString(self) -> str:
        transports = {
            0: "CELLULAR", 1: "WIFI", 2: "BT", 3: "ETHERNET",
            4: "VPN", 5: "WIFI_AWARE",
        }
        trans_str = ", ".join(transports.get(t, str(t)) for t in sorted(self._transport_types))
        return f"NetworkCapabilities(transports=[{trans_str}], caps=[{self.describeCapabilities()}])"

    def equals(self, other: Any) -> bool:
        if not isinstance(other, NetworkCapabilities):
            return False
        return (self._transport_types == other._transport_types and
                self._capabilities == other._capabilities)

    def hashCode(self) -> int:
        return hash((frozenset(self._transport_types), frozenset(self._capabilities))) & 0x7FFFFFFF


# ═══════════════════════════════════════════════════════════════
# NetworkSpecifier / MatchAllNetworkSpecifier
# ═══════════════════════════════════════════════════════════════

class NetworkSpecifier(Stub):
    """NetworkSpecifier — 网络指定器"""

    def satisfiedBy(self, other: Optional["NetworkSpecifier"]) -> bool:
        return False

    def toString(self) -> str:
        return f"NetworkSpecifier@{hex(id(self))}"

    def equals(self, other: Any) -> bool:
        return self is other

    def hashCode(self) -> int:
        return id(self) & 0x7FFFFFFF


class MatchAllNetworkSpecifier(NetworkSpecifier):
    """MatchAllNetworkSpecifier — 匹配所有网络"""

    def satisfiedBy(self, other: Optional["NetworkSpecifier"]) -> bool:
        return True

    def toString(self) -> str:
        return "MatchAllNetworkSpecifier"


# ═══════════════════════════════════════════════════════════════
# NetworkRequest
# ═══════════════════════════════════════════════════════════════

class NetworkRequest(Stub):
    """NetworkRequest — 网络请求"""

    def __init__(self, capabilities: Optional[NetworkCapabilities] = None,
                 networkSpecifier: Optional[NetworkSpecifier] = None,
                 legacyType: int = -1, isRequest: bool = True):
        super().__init__()
        self._capabilities = capabilities or NetworkCapabilities()
        self._network_specifier = networkSpecifier
        self._legacy_type = legacyType
        self._is_request = isRequest

    def getCapabilities(self) -> NetworkCapabilities:
        return self._capabilities

    def getNetworkSpecifier(self) -> Optional[NetworkSpecifier]:
        return self._network_specifier

    def getLegacyType(self) -> int:
        return self._legacy_type

    def hasCapability(self, capability: int) -> bool:
        return self._capabilities.hasCapability(capability)

    def hasTransport(self, transport: int) -> bool:
        return self._capabilities.hasTransport(transport)

    class Builder(Stub):
        """NetworkRequest.Builder — 请求构建器"""

        def __init__(self):
            super().__init__()
            self._capabilities = NetworkCapabilities()
            self._network_specifier: Optional[NetworkSpecifier] = None

        def addTransportType(self, transportType: int) -> "NetworkRequest.Builder":
            self._capabilities.addTransportType(transportType)
            return self

        def removeTransportType(self, transportType: int) -> "NetworkRequest.Builder":
            self._capabilities.removeTransportType(transportType)
            return self

        def addCapability(self, capability: int) -> "NetworkRequest.Builder":
            self._capabilities.addCapability(capability)
            return self

        def removeCapability(self, capability: int) -> "NetworkRequest.Builder":
            self._capabilities.removeCapability(capability)
            return self

        def addCapabilities(self, capabilities: List[int]) -> "NetworkRequest.Builder":
            for c in capabilities:
                self._capabilities.addCapability(c)
            return self

        def removeCapabilities(self, capabilities: List[int]) -> "NetworkRequest.Builder":
            for c in capabilities:
                self._capabilities.removeCapability(c)
            return self

        def setNetworkSpecifier(self, specifier: NetworkSpecifier) -> "NetworkRequest.Builder":
            self._network_specifier = specifier
            return self

        def build(self) -> "NetworkRequest":
            caps = self._capabilities
            caps.setNetworkSpecifier(self._network_specifier)
            return NetworkRequest(capabilities=caps, networkSpecifier=self._network_specifier)

    def toString(self) -> str:
        return f"NetworkRequest(caps={self._capabilities})"

    def equals(self, other: Any) -> bool:
        if not isinstance(other, NetworkRequest):
            return False
        return self._capabilities == other._capabilities

    def hashCode(self) -> int:
        return self._capabilities.hashCode()


# ═══════════════════════════════════════════════════════════════
# NetworkAgent
# ═══════════════════════════════════════════════════════════════

class NetworkAgent(Stub):
    """NetworkAgent — 网络代理"""

    def __init__(self, looper: Any = None, network: Optional[Network] = None,
                 capabilities: Optional[NetworkCapabilities] = None,
                 linkProperties: Optional["LinkProperties"] = None,
                 score: int = 0):
        super().__init__()
        self._network = network or Network()
        self._capabilities = capabilities or NetworkCapabilities()
        self._link_properties = linkProperties or LinkProperties()
        self._score = score
        self._validated = False
        self._connected = False
        self._callback: Optional[Callable] = None

    def getNetwork(self) -> Network:
        return self._network

    def getCapabilities(self) -> NetworkCapabilities:
        return self._capabilities

    def getLinkProperties(self) -> "LinkProperties":
        return self._link_properties

    def getScore(self) -> int:
        return self._score

    def setScore(self, score: int):
        self._score = score

    def connect(self):
        self._connected = True

    def disconnect(self):
        self._connected = False

    def isConnected(self) -> bool:
        return self._connected

    def isValidated(self) -> bool:
        return self._validated

    def setValidated(self, validated: bool):
        self._validated = validated

    def sendNetworkCapabilities(self, capabilities: NetworkCapabilities):
        self._capabilities = capabilities

    def sendLinkProperties(self, linkProperties: "LinkProperties"):
        self._link_properties = linkProperties

    def sendNetworkScore(self, score: int):
        self._score = score

    def registerCallback(self, callback: Callable):
        self._callback = callback

    def unregisterCallback(self, callback: Callable):
        if self._callback is callback:
            self._callback = None

    def toString(self) -> str:
        return f"NetworkAgent(net={self._network}, score={self._score}, connected={self._connected})"

    def equals(self, other: Any) -> bool:
        if not isinstance(other, NetworkAgent):
            return False
        return self._network == other._network

    def hashCode(self) -> int:
        return self._network.hashCode()


# ═══════════════════════════════════════════════════════════════
# NetworkInfo (扩展)
# ═══════════════════════════════════════════════════════════════

class NetworkInfo(Stub):
    """NetworkInfo — 网络信息 (完整版)"""

    TYPE_NONE = -1
    TYPE_MOBILE = 0
    TYPE_WIFI = 1
    TYPE_WIMAX = 6
    TYPE_ETHERNET = 9
    TYPE_BLUETOOTH = 7
    TYPE_VPN = 17
    TYPE_WIFI_P2P = 13
    TYPE_MOBILE_MMS = 2
    TYPE_MOBILE_SUPL = 3
    TYPE_MOBILE_DUN = 4
    TYPE_MOBILE_HIPRI = 5
    TYPE_MOBILE_FOTA = 10
    TYPE_MOBILE_IMS = 11
    TYPE_MOBILE_CBS = 12
    TYPE_MOBILE_IA = 14
    TYPE_MOBILE_EMERGENCY = 15
    TYPE_PROXY = 16

    STATE_CONNECTING = 0
    STATE_CONNECTED = 1
    STATE_SUSPENDED = 2
    STATE_DISCONNECTING = 3
    STATE_DISCONNECTED = 4
    STATE_UNKNOWN = 5

    DETAILED_STATE_IDLE = 0
    DETAILED_STATE_SCANNING = 1
    DETAILED_STATE_CONNECTING = 2
    DETAILED_STATE_AUTHENTICATING = 3
    DETAILED_STATE_OBTAINING_IPADDR = 4
    DETAILED_STATE_CONNECTED = 5
    DETAILED_STATE_SUSPENDED = 6
    DETAILED_STATE_DISCONNECTING = 7
    DETAILED_STATE_DISCONNECTED = 8
    DETAILED_STATE_FAILED = 9
    DETAILED_STATE_BLOCKED = 10
    DETAILED_STATE_VERIFYING_POOR_LINK = 11
    DETAILED_STATE_CAPTIVE_PORTAL_CHECK = 12

    def __init__(self, type_: int = TYPE_NONE, subtype: int = 0,
                 state: int = STATE_DISCONNECTED,
                 detailedState: int = DETAILED_STATE_IDLE,
                 isAvailable: bool = True,
                 isRoaming: bool = False,
                 extraInfo: str = "",
                 isFailover: bool = False,
                 reason: str = ""):
        super().__init__()
        self._type = type_
        self._subtype = subtype
        self._state = state
        self._detailed_state = detailedState
        self._is_available = isAvailable
        self._is_roaming = isRoaming
        self._extra_info = extraInfo
        self._is_failover = isFailover
        self._reason = reason

    def getType(self) -> int:
        return self._type

    def getSubtype(self) -> int:
        return self._subtype

    def getState(self) -> int:
        return self._state

    def getDetailedState(self) -> int:
        return self._detailed_state

    def isConnected(self) -> bool:
        return self._state == NetworkInfo.STATE_CONNECTED

    def isConnectedOrConnecting(self) -> bool:
        return self._state in (NetworkInfo.STATE_CONNECTED, NetworkInfo.STATE_CONNECTING)

    def isAvailable(self) -> bool:
        return self._is_available

    def isRoaming(self) -> bool:
        return self._is_roaming

    def isFailover(self) -> bool:
        return self._is_failover

    def getExtraInfo(self) -> str:
        return self._extra_info

    def getReason(self) -> str:
        return self._reason

    def getTypeName(self) -> str:
        names = {
            -1: "NONE", 0: "MOBILE", 1: "WIFI", 6: "WIMAX",
            9: "ETHERNET", 7: "BLUETOOTH", 17: "VPN",
            13: "WIFI_P2P", 16: "PROXY",
        }
        return names.get(self._type, f"TYPE_{self._type}")

    def setDetailedState(self, detailedState: int, reason: str, extraInfo: str):
        self._detailed_state = detailedState
        self._reason = reason
        self._extra_info = extraInfo

    def setState(self, state: int):
        self._state = state

    def toString(self) -> str:
        return (f"NetworkInfo(type={self.getTypeName()}, state={self._state}, "
                f"connected={self.isConnected()})")

    def equals(self, other: Any) -> bool:
        if not isinstance(other, NetworkInfo):
            return False
        return self._type == other._type and self._state == other._state

    def hashCode(self) -> int:
        return hash((self._type, self._state)) & 0x7FFFFFFF


# ═══════════════════════════════════════════════════════════════
# LinkAddress
# ═══════════════════════════════════════════════════════════════

class LinkAddress(Stub):
    """LinkAddress — 链路地址"""

    def __init__(self, address: str = "0.0.0.0", prefixLength: int = 24,
                 flags: int = 0, scope: int = 0):
        super().__init__()
        self._address = address
        self._prefix_length = prefixLength
        self._flags = flags
        self._scope = scope

    def getAddress(self) -> str:
        return self._address

    def getPrefixLength(self) -> int:
        return self._prefix_length

    def getFlags(self) -> int:
        return self._flags

    def getScope(self) -> int:
        return self._scope

    def isGlobalPreferred(self) -> bool:
        return self._scope == 0

    def toString(self) -> str:
        return f"{self._address}/{self._prefix_length}"

    def equals(self, other: Any) -> bool:
        if not isinstance(other, LinkAddress):
            return False
        return self._address == other._address and self._prefix_length == other._prefix_length

    def hashCode(self) -> int:
        return hash((self._address, self._prefix_length)) & 0x7FFFFFFF


# ═══════════════════════════════════════════════════════════════
# LinkProperties
# ═══════════════════════════════════════════════════════════════

class LinkProperties(Stub):
    """LinkProperties — 链路属性"""

    def __init__(self):
        super().__init__()
        self._interface_name: str = ""
        self._link_addresses: List[LinkAddress] = []
        self._dns_servers: List[str] = []
        self._domains: str = ""
        self._mtu: int = 1500
        self._http_proxy: Optional["ProxyInfo"] = None
        self._private_dns_server_name: str = ""
        self._routes: List[str] = []

    def setInterfaceName(self, name: str):
        self._interface_name = name

    def getInterfaceName(self) -> str:
        return self._interface_name

    def addLinkAddress(self, address: LinkAddress):
        self._link_addresses.append(address)

    def removeLinkAddress(self, address: LinkAddress):
        self._link_addresses = [a for a in self._link_addresses if a != address]

    def getLinkAddresses(self) -> List[LinkAddress]:
        return list(self._link_addresses)

    def addDnsServer(self, dnsServer: str):
        self._dns_servers.append(dnsServer)

    def removeDnsServer(self, dnsServer: str):
        self._dns_servers = [d for d in self._dns_servers if d != dnsServer]

    def getDnsServers(self) -> List[str]:
        return list(self._dns_servers)

    def setDomains(self, domains: str):
        self._domains = domains

    def getDomains(self) -> str:
        return self._domains

    def setMtu(self, mtu: int):
        self._mtu = mtu

    def getMtu(self) -> int:
        return self._mtu

    def setHttpProxy(self, proxy: "ProxyInfo"):
        self._http_proxy = proxy

    def getHttpProxy(self) -> Optional["ProxyInfo"]:
        return self._http_proxy

    def setPrivateDnsServerName(self, name: str):
        self._private_dns_server_name = name

    def getPrivateDnsServerName(self) -> str:
        return self._private_dns_server_name

    def addRoute(self, route: str):
        self._routes.append(route)

    def removeRoute(self, route: str):
        self._routes = [r for r in self._routes if r != route]

    def getRoutes(self) -> List[str]:
        return list(self._routes)

    def clear(self):
        self._link_addresses.clear()
        self._dns_servers.clear()
        self._routes.clear()

    def toString(self) -> str:
        return (f"LinkProperties(iface={self._interface_name}, "
                f"dns={self._dns_servers}, mtu={self._mtu})")

    def equals(self, other: Any) -> bool:
        if not isinstance(other, LinkProperties):
            return False
        return self._interface_name == other._interface_name

    def hashCode(self) -> int:
        return hash(self._interface_name) & 0x7FFFFFFF


# ═══════════════════════════════════════════════════════════════
# LinkQualityInfo
# ═══════════════════════════════════════════════════════════════

class LinkQualityInfo(Stub):
    """LinkQualityInfo — 链路质量信息"""

    NORMALIZED_MIN = 0
    NORMALIZED_MAX = 100

    def __init__(self):
        super().__init__()
        self._normalized_rssi = LinkQualityInfo.NORMALIZED_MIN
        self._normalized_signal_strength = LinkQualityInfo.NORMALIZED_MIN
        self._timestamp = 0
        self._theoretical_tx_rate = 0
        self._theoretical_rx_rate = 0

    def getNormalizedRssi(self) -> int:
        return self._normalized_rssi

    def setNormalizedRssi(self, rssi: int):
        self._normalized_rssi = max(LinkQualityInfo.NORMALIZED_MIN,
                                     min(LinkQualityInfo.NORMALIZED_MAX, rssi))

    def getNormalizedSignalStrength(self) -> int:
        return self._normalized_signal_strength

    def setNormalizedSignalStrength(self, strength: int):
        self._normalized_signal_strength = max(LinkQualityInfo.NORMALIZED_MIN,
                                                min(LinkQualityInfo.NORMALIZED_MAX, strength))

    def getTimestamp(self) -> int:
        return self._timestamp

    def setTimestamp(self, timestamp: int):
        self._timestamp = timestamp

    def getTheoreticalTxRate(self) -> int:
        return self._theoretical_tx_rate

    def setTheoreticalTxRate(self, rate: int):
        self._theoretical_tx_rate = rate

    def getTheoreticalRxRate(self) -> int:
        return self._theoretical_rx_rate

    def setTheoreticalRxRate(self, rate: int):
        self._theoretical_rx_rate = rate

    def toString(self) -> str:
        return f"LinkQualityInfo(rssi={self._normalized_rssi}, txRate={self._theoretical_tx_rate})"

    def equals(self, other: Any) -> bool:
        return self is other

    def hashCode(self) -> int:
        return id(self) & 0x7FFFFFFF


class WifiLinkQualityInfo(LinkQualityInfo):
    """WifiLinkQualityInfo — WiFi 链路质量"""

    def __init__(self):
        super().__init__()
        self._ssid: str = ""
        self._bssid: str = ""
        self._rssi: int = 0
        self._link_speed_mbps: int = 0
        self._frequency: int = 0

    def getSsid(self) -> str:
        return self._ssid

    def setSsid(self, ssid: str):
        self._ssid = ssid

    def getBssid(self) -> str:
        return self._bssid

    def setBssid(self, bssid: str):
        self._bssid = bssid

    def getRssi(self) -> int:
        return self._rssi

    def setRssi(self, rssi: int):
        self._rssi = rssi

    def getLinkSpeedMbps(self) -> int:
        return self._link_speed_mbps

    def setLinkSpeedMbps(self, speed: int):
        self._link_speed_mbps = speed

    def getFrequency(self) -> int:
        return self._frequency

    def setFrequency(self, freq: int):
        self._frequency = freq

    def toString(self) -> str:
        return f"WifiLinkQualityInfo(ssid={self._ssid}, rssi={self._rssi})"


class EthernetLinkQualityInfo(LinkQualityInfo):
    """EthernetLinkQualityInfo — 以太网链路质量"""

    def __init__(self):
        super().__init__()
        self._cable_status: int = 0
        self._link_speed: int = 1000

    def getCableStatus(self) -> int:
        return self._cable_status

    def setCableStatus(self, status: int):
        self._cable_status = status

    def getLinkSpeed(self) -> int:
        return self._link_speed

    def setLinkSpeed(self, speed: int):
        self._link_speed = speed

    def toString(self) -> str:
        return f"EthernetLinkQualityInfo(speed={self._link_speed})"


class MobileLinkQualityInfo(LinkQualityInfo):
    """MobileLinkQualityInfo — 移动网络链路质量"""

    def __init__(self):
        super().__init__()
        self._network_type: int = 0
        self._signal_dbm: int = 0
        self._sinr: int = 0
        self._rsrp: int = 0
        self._rsrq: int = 0
        self._rssnr: int = 0

    def getNetworkType(self) -> int:
        return self._network_type

    def setNetworkType(self, type_: int):
        self._network_type = type_

    def getSignalDbm(self) -> int:
        return self._signal_dbm

    def setSignalDbm(self, dbm: int):
        self._signal_dbm = dbm

    def getSinr(self) -> int:
        return self._sinr

    def setSinr(self, sinr: int):
        self._sinr = sinr

    def getRsrp(self) -> int:
        return self._rsrp

    def setRsrp(self, rsrp: int):
        self._rsrp = rsrp

    def getRsrq(self) -> int:
        return self._rsrq

    def setRsrq(self, rsrq: int):
        self._rsrq = rsrq

    def getRssnr(self) -> int:
        return self._rssnr

    def setRssnr(self, rssnr: int):
        self._rssnr = rssnr

    def toString(self) -> str:
        return f"MobileLinkQualityInfo(type={self._network_type}, rsrp={self._rsrp})"


# ═══════════════════════════════════════════════════════════════
# TrafficStats
# ═══════════════════════════════════════════════════════════════

class TrafficStats(Stub):
    """TrafficStats — 流量统计"""

    UID_REMOVED = -4
    UID_TETHERING = -5
    UNSUPPORTED = -1

    @staticmethod
    def getMobileTxPackets() -> int:
        return 0

    @staticmethod
    def getMobileRxPackets() -> int:
        return 0

    @staticmethod
    def getMobileTxBytes() -> int:
        return 0

    @staticmethod
    def getMobileRxBytes() -> int:
        return 0

    @staticmethod
    def getTotalTxPackets() -> int:
        return 0

    @staticmethod
    def getTotalRxPackets() -> int:
        return 0

    @staticmethod
    def getTotalTxBytes() -> int:
        return 0

    @staticmethod
    def getTotalRxBytes() -> int:
        return 0

    @staticmethod
    def getUidTxBytes(uid: int) -> int:
        return 0

    @staticmethod
    def getUidRxBytes(uid: int) -> int:
        return 0

    @staticmethod
    def getUidTxPackets(uid: int) -> int:
        return 0

    @staticmethod
    def getUidRxPackets(uid: int) -> int:
        return 0

    @staticmethod
    def getUidTcpTxBytes(uid: int) -> int:
        return 0

    @staticmethod
    def getUidTcpRxBytes(uid: int) -> int:
        return 0

    @staticmethod
    def getUidUdpTxBytes(uid: int) -> int:
        return 0

    @staticmethod
    def getUidUdpRxBytes(uid: int) -> int:
        return 0

    @staticmethod
    def getUidTcpSegmentsSent(uid: int) -> int:
        return 0

    @staticmethod
    def getUidTcpSegmentsReceived(uid: int) -> int:
        return 0

    @staticmethod
    def getUidUdpPacketsSent(uid: int) -> int:
        return 0

    @staticmethod
    def getUidUdpPacketsReceived(uid: int) -> int:
        return 0

    def toString(self) -> str:
        return "TrafficStats(desktop stub)"


# ═══════════════════════════════════════════════════════════════
# ProxyInfo
# ═══════════════════════════════════════════════════════════════

class ProxyInfo(Stub):
    """ProxyInfo — 代理信息"""

    def __init__(self, host: str = "", port: int = 0,
                 exclusionList: List[str] = None,
                 pacFileUrl: str = ""):
        super().__init__()
        self._host = host
        self._port = port
        self._exclusion_list = exclusion_list or []
        self._pac_file_url = pacFileUrl

    def getHost(self) -> str:
        return self._host

    def getPort(self) -> int:
        return self._port

    def getExclusionList(self) -> List[str]:
        return list(self._exclusion_list)

    def getPacFileUrl(self) -> str:
        return self._pac_file_url

    def isValid(self) -> bool:
        return bool(self._host) and self._port > 0

    def toString(self) -> str:
        return f"ProxyInfo(host={self._host}, port={self._port})"

    def equals(self, other: Any) -> bool:
        if not isinstance(other, ProxyInfo):
            return False
        return self._host == other._host and self._port == other._port

    def hashCode(self) -> int:
        return hash((self._host, self._port)) & 0x7FFFFFFF


# ═══════════════════════════════════════════════════════════════
# DhcpInfo / DhcpInfoInternal
# ═══════════════════════════════════════════════════════════════

class DhcpInfo(Stub):
    """DhcpInfo — DHCP 信息"""

    def __init__(self):
        super().__init__()
        self.ipAddress: int = 0
        self.gateway: int = 0
        self.netmask: int = 0
        self.dns1: int = 0
        self.dns2: int = 0
        self.serverAddress: int = 0
        self.leaseDuration: int = 0

    def toString(self) -> str:
        def int_to_ip(ip: int) -> str:
            return ".".join(str((ip >> (8 * i)) & 0xFF) for i in range(4))
        return (f"DhcpInfo(ip={int_to_ip(self.ipAddress)}, "
                f"gateway={int_to_ip(self.gateway)}, "
                f"dns1={int_to_ip(self.dns1)})")

    def equals(self, other: Any) -> bool:
        if not isinstance(other, DhcpInfo):
            return False
        return self.ipAddress == other.ipAddress

    def hashCode(self) -> int:
        return self.ipAddress & 0x7FFFFFFF


class DhcpInfoInternal(Stub):
    """DhcpInfoInternal — DHCP 内部信息"""

    def __init__(self):
        super().__init__()
        self.ipAddress: str = ""
        self.gateway: str = ""
        self.netmask: str = ""
        self.dns1: str = ""
        self.dns2: str = ""
        self.serverAddress: str = ""
        self.leaseDuration: int = 0
        self.domain: str = ""

    def toString(self) -> str:
        return f"DhcpInfoInternal(ip={self.ipAddress}, gateway={self.gateway})"

    def equals(self, other: Any) -> bool:
        if not isinstance(other, DhcpInfoInternal):
            return False
        return self.ipAddress == other.ipAddress

    def hashCode(self) -> int:
        return hash(self.ipAddress) & 0x7FFFFFFF


# ═══════════════════════════════════════════════════════════════
# NetworkBadging
# ═══════════════════════════════════════════════════════════════

class NetworkBadging(Stub):
    """NetworkBadging — 网络徽章"""

    BADGING_WIFI = 0
    BADGING_4G = 1
    BADGING_3G = 2
    BADGING_2G = 3
    BADGING_1X = 4
    BADGING_EHRPD = 5
    BADGING_HSPAP = 6
    BADGING_LTE = 7
    BADGING_LTE_CA = 8
    BADGING_NR = 9
    BADGING_ETHERNET = 10
    BADGING_WIMAX = 11

    @staticmethod
    def getBadging(transportType: int, subType: int = 0) -> int:
        mapping = {1: NetworkBadging.BADGING_WIFI, 3: NetworkBadging.BADGING_ETHERNET}
        return mapping.get(transportType, NetworkBadging.BADGING_4G)

    def toString(self) -> str:
        return f"NetworkBadging@{hex(id(self))}"


# ═══════════════════════════════════════════════════════════════
# ConnectivityDiagnosticsManager
# ═══════════════════════════════════════════════════════════════

class ConnectivityDiagnosticsManager(Stub):
    """ConnectivityDiagnosticsManager — 连接诊断管理器"""

    def __init__(self):
        super().__init__()
        self._callbacks: Dict[int, Callable] = {}
        self._reports: Dict[int, "ConnectivityDiagnosticsManager_ConnectivityReport"] = {}
        self._stall_reports: Dict[int, "ConnectivityDiagnosticsManager_DataStallReport"] = {}

    class ConnectivityReport(Stub):
        """ConnectivityDiagnosticsManager.ConnectivityReport — 连接报告"""

        def __init__(self, network: Optional[Network] = None,
                     linkProperties: Optional[LinkProperties] = None,
                     networkCapabilities: Optional[NetworkCapabilities] = None,
                     timestamp: int = 0,
                     errorCode: int = 0,
                     errorMessage: str = ""):
            super().__init__()
            self._network = network
            self._link_properties = linkProperties
            self._network_capabilities = networkCapabilities
            self._timestamp = timestamp
            self._error_code = errorCode
            self._error_message = errorMessage

        def getNetwork(self) -> Optional[Network]:
            return self._network

        def getLinkProperties(self) -> Optional[LinkProperties]:
            return self._link_properties

        def getNetworkCapabilities(self) -> Optional[NetworkCapabilities]:
            return self._network_capabilities

        def getTimestamp(self) -> int:
            return self._timestamp

        def getErrorCode(self) -> int:
            return self._error_code

        def getErrorMessage(self) -> str:
            return self._error_message

        def toString(self) -> str:
            return f"ConnectivityReport(net={self._network}, error={self._error_code})"

        def equals(self, other: Any) -> bool:
            if not isinstance(other, ConnectivityDiagnosticsManager.ConnectivityReport):
                return False
            return self._network == other._network and self._timestamp == other._timestamp

        def hashCode(self) -> int:
            return hash((id(self._network), self._timestamp)) & 0x7FFFFFFF

    class DataStallReport(Stub):
        """ConnectivityDiagnosticsManager.DataStallReport — 数据卡顿报告"""

        REASON_NETWORK_VALIDATION = 0
        REASON_NETWORK_CONGESTION = 1
        REASON_NETWORK_PARTIAL_CONNECTIVITY = 2
        REASON_NETWORK_SUSPECTED = 3
        REASON_DNS_RESOLUTION_FAILURE = 4
        REASON_TCP_FAILURE = 5
        REASON_IP_REACHABILITY_LOSS = 6

        DETECTION_METHOD_NONE = 0
        DETECTION_METHOD_DNS_EVENTS = 1
        DETECTION_METHOD_TCP_METRICS = 2
        DETECTION_METHOD_IP_METRICS = 3

        def __init__(self, network: Optional[Network] = None,
                     detectionMethod: int = DETECTION_METHOD_NONE,
                     reason: int = REASON_NETWORK_VALIDATION,
                     timestamp: int = 0,
                     linkProperties: Optional[LinkProperties] = None,
                     networkCapabilities: Optional[NetworkCapabilities] = None):
            super().__init__()
            self._network = network
            self._detection_method = detectionMethod
            self._reason = reason
            self._timestamp = timestamp
            self._link_properties = linkProperties
            self._network_capabilities = networkCapabilities

        def getNetwork(self) -> Optional[Network]:
            return self._network

        def getDetectionMethod(self) -> int:
            return self._detection_method

        def getReason(self) -> int:
            return self._reason

        def getTimestamp(self) -> int:
            return self._timestamp

        def getLinkProperties(self) -> Optional[LinkProperties]:
            return self._link_properties

        def getNetworkCapabilities(self) -> Optional[NetworkCapabilities]:
            return self._network_capabilities

        def toString(self) -> str:
            return f"DataStallReport(net={self._network}, reason={self._reason})"

        def equals(self, other: Any) -> bool:
            if not isinstance(other, ConnectivityDiagnosticsManager.DataStallReport):
                return False
            return self._network == other._network and self._timestamp == other._timestamp

        def hashCode(self) -> int:
            return hash((id(self._network), self._timestamp)) & 0x7FFFFFFF

    def registerConnectivityDiagnosticsCallback(
            self, request: NetworkRequest,
            callback: Callable,
            handler: Any = None):
        lid = id(callback)
        self._callbacks[lid] = callback

    def unregisterConnectivityDiagnosticsCallback(self, callback: Callable):
        lid = id(callback)
        self._callbacks.pop(lid, None)

    def getConnectivityReport(self, network: Network) -> Optional[ConnectivityReport]:
        return self._reports.get(network.getNetId())

    def getDataStallReport(self, network: Network) -> Optional[DataStallReport]:
        return self._stall_reports.get(network.getNetId())

    def toString(self) -> str:
        return f"ConnectivityDiagnosticsManager(callbacks={len(self._callbacks)})"

    def equals(self, other: Any) -> bool:
        return self is other

    def hashCode(self) -> int:
        return id(self) & 0x7FFFFFFF


# ═══════════════════════════════════════════════════════════════
# ConnectivityManager (扩展)
# ═══════════════════════════════════════════════════════════════

class ConnectivityManager(Stub):
    """ConnectivityManager — 连接管理器 (扩展版)"""

    TYPE_NONE = -1
    TYPE_MOBILE = 0
    TYPE_WIFI = 1
    TYPE_MOBILE_MMS = 2
    TYPE_MOBILE_SUPL = 3
    TYPE_MOBILE_DUN = 4
    TYPE_MOBILE_HIPRI = 5
    TYPE_WIMAX = 6
    TYPE_BLUETOOTH = 7
    TYPE_DUMMY = 8
    TYPE_ETHERNET = 9
    TYPE_MOBILE_FOTA = 10
    TYPE_MOBILE_IMS = 11
    TYPE_MOBILE_CBS = 12
    TYPE_WIFI_P2P = 13
    TYPE_MOBILE_IA = 14
    TYPE_MOBILE_EMERGENCY = 15
    TYPE_PROXY = 16
    TYPE_VPN = 17

    EXTRA_NETWORK = "android.net.extra.NETWORK"
    EXTRA_OTHER_NETWORK = "android.net.extra.OTHER_NETWORK"
    EXTRA_NO_CONNECTIVITY = "android.net.extra.NO_CONNECTIVITY"
    EXTRA_REASON = "android.net.extra.REASON"
    EXTRA_EXTRA_INFO = "android.net.extra.EXTRA_INFO"
    EXTRA_NETWORK_INFO = "android.net.extra.NETWORK_INFO"
    EXTRA_IS_FAILOVER = "android.net.extra.IS_FAILOVER"
    EXTRA_NETWORK_TYPE = "android.net.extra.NETWORK_TYPE"

    ACTION_CAPTIVE_PORTAL_SIGN_IN = "android.net.conn.CAPTIVE_PORTAL"
    ACTION_CAPTIVE_PORTAL_TEST_COMPLETED = "android.net.conn.CAPTIVE_PORTAL_TEST_COMPLETED"

    def __init__(self):
        super().__init__()
        self._active_network: Optional[Network] = None
        self._networks: Dict[int, Network] = {}
        self._network_infos: Dict[int, NetworkInfo] = {}
        self._callbacks: Dict[int, Any] = {}
        self._diagnostics_manager = ConnectivityDiagnosticsManager()

    def getActiveNetwork(self) -> Optional[Network]:
        return self._active_network

    def _get_linux_active_info(self):
        """从 Linux 真实网络信息构造活跃 NetworkInfo"""
        try:
            from .net import NetworkInfo
            interfaces = LinuxNetwork.get_interfaces()
            # 有真实网络接口 (非环回) 且 operstate=up 即视为已连接
            connected = any(
                (iface.get("ifname") or iface.get("name", "")) != "lo"
                and iface.get("operstate", "").lower() == "up"
                for iface in interfaces
            )
            if not connected:
                connected = LinuxNetwork.get_connectivity()
            ip = LinuxNetwork.get_ip_address()
            iface_name = ""
            net_type = ConnectivityManager.TYPE_ETHERNET
            for iface in interfaces:
                name = iface.get("ifname") or iface.get("name", "")
                state = iface.get("operstate", "")
                if name and name != "lo" and state.lower() == "up":
                    iface_name = iface_name or name
                if name and name.startswith(("wl", "wlan", "wlp")):
                    net_type = ConnectivityManager.TYPE_WIFI
                    iface_name = name
            data = {
                "connected": connected,
                "ip": ip,
                "interface": iface_name,
                "type_name": "WIFI" if net_type == ConnectivityManager.TYPE_WIFI else "ETHERNET",
            }
            return NetworkInfo.from_linux_data(data, network_type=net_type)
        except Exception:
            return None

    def getActiveNetworkInfo(self) -> Optional[NetworkInfo]:
        if _USE_LINUX:
            return self._get_linux_active_info()
        if self._active_network:
            return self._network_infos.get(self._active_network.getNetId())
        return None

    def getNetworkInfo(self, networkType: int) -> Optional[NetworkInfo]:
        if _USE_LINUX:
            info = self._get_linux_active_info()
            if info is not None and info.getType() == networkType:
                return info
            return None
        for info in self._network_infos.values():
            if info.getType() == networkType:
                return info
        return None

    def getConnectivityStatus(self) -> bool:
        """返回当前网络连通性 (True=已连接)"""
        if _USE_LINUX:
            return LinuxNetwork.get_connectivity()
        return self._active_network is not None

    def getNetworkInfoForNetwork(self, network: Network) -> Optional[NetworkInfo]:
        return self._network_infos.get(network.getNetId())

    def getAllNetworks(self) -> List[Network]:
        return list(self._networks.values())

    def getAllNetworkInfo(self) -> List[NetworkInfo]:
        return list(self._network_infos.values())

    def getNetworkCapabilities(self, network: Network) -> Optional[NetworkCapabilities]:
        return network.getNetworkCapabilities()

    def getLinkProperties(self, network: Network) -> Optional[LinkProperties]:
        return network.getLinkProperties()

    def requestNetwork(self, request: NetworkRequest, callback: Any,
                       handler: Any = None, timeoutMs: int = 0,
                       flags: int = 0) -> "NetworkRequest":
        lid = id(callback)
        self._callbacks[lid] = callback
        return request

    def unregisterNetworkCallback(self, callback: Any):
        lid = id(callback)
        self._callbacks.pop(lid, None)

    def registerNetworkCallback(self, request: NetworkRequest, callback: Any,
                                handler: Any = None):
        lid = id(callback)
        self._callbacks[lid] = callback

    def registerDefaultNetworkCallback(self, callback: Any, handler: Any = None):
        lid = id(callback)
        self._callbacks[lid] = callback

    def bindProcessToNetwork(self, network: Network) -> bool:
        return True

    def getProcessDefaultNetwork(self) -> Optional[Network]:
        return self._active_network

    def setProcessDefaultNetwork(self, network: Optional[Network]) -> bool:
        self._active_network = network
        return True

    def isDefaultNetworkActive(self) -> bool:
        return self._active_network is not None

    def addDefaultNetworkActiveListener(self, callback: Callable):
        pass

    def removeDefaultNetworkActiveListener(self, callback: Callable):
        pass

    def reportInetCondition(self, networkType: int, percentage: int):
        pass

    def getActiveLinkQualityInfo(self) -> Optional[LinkQualityInfo]:
        return None

    def getLinkQualityInfo(self, networkType: int) -> Optional[LinkQualityInfo]:
        return None

    def getAllLinkQualityInfo(self) -> List[LinkQualityInfo]:
        return []

    def getConnectivityDiagnosticsManager(self) -> ConnectivityDiagnosticsManager:
        return self._diagnostics_manager

    def toString(self) -> str:
        return f"ConnectivityManager(active={self._active_network})"

    def equals(self, other: Any) -> bool:
        return self is other

    def hashCode(self) -> int:
        return id(self) & 0x7FFFFFFF