""" 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 from .net import ConnectivityManager as _BaseConnectivityManager from .net import DhcpInfo as _BaseDhcpInfo from .net import LinkAddress as _BaseLinkAddress from .net import LinkProperties as _BaseLinkProperties from .net import Network as _BaseNetwork from .net import NetworkCapabilities as _BaseNetworkCapabilities from .net import NetworkInfo as _BaseNetworkInfo from .net import NetworkRequest as _BaseNetworkRequest from .net import NetworkSpecifier as _BaseNetworkSpecifier from .net import ProxyInfo as _BaseProxyInfo from .net import TrafficStats as _BaseTrafficStats from .net import MatchAllNetworkSpecifier as _BaseMatchAllNetworkSpecifier # 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(_BaseNetwork): """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(_BaseNetworkCapabilities): """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(_BaseNetworkSpecifier): """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(_BaseMatchAllNetworkSpecifier, NetworkSpecifier): """MatchAllNetworkSpecifier — 匹配所有网络""" def satisfiedBy(self, other: Optional["NetworkSpecifier"]) -> bool: return True def toString(self) -> str: return "MatchAllNetworkSpecifier" # ═══════════════════════════════════════════════════════════════ # NetworkRequest # ═══════════════════════════════════════════════════════════════ class NetworkRequest(_BaseNetworkRequest): """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(_BaseNetworkInfo): """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(_BaseLinkAddress): """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(_BaseLinkProperties): """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(_BaseTrafficStats): """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(_BaseProxyInfo): """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(_BaseDhcpInfo): """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(_BaseConnectivityManager): """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