304 lines
13 KiB
Python
304 lines
13 KiB
Python
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from typing import Callable, Optional
|
|
|
|
from core.Function.common import preferred_encoding, prefix_to_netmask, run_hidden, validate_ip
|
|
|
|
|
|
OutputCallback = Callable[[str, Optional[str]], None]
|
|
|
|
|
|
class NetworkManager:
|
|
def __init__(self, output: OutputCallback):
|
|
self.output = output
|
|
self.profiles_file = os.path.join(self._base_dir(), "network_profiles.json")
|
|
|
|
def _base_dir(self) -> str:
|
|
if getattr(sys, "frozen", False):
|
|
return os.path.dirname(sys.executable)
|
|
return os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
def get_network_info(self) -> list[dict]:
|
|
script = r"""
|
|
$ErrorActionPreference = "SilentlyContinue"
|
|
$items = Get-NetIPConfiguration | ForEach-Object {
|
|
$alias = $_.InterfaceAlias
|
|
$index = $_.InterfaceIndex
|
|
$adapter = Get-NetAdapter -InterfaceAlias $alias -ErrorAction SilentlyContinue
|
|
$ipif = Get-NetIPInterface -InterfaceAlias $alias -AddressFamily IPv4 -ErrorAction SilentlyContinue | Select-Object -First 1
|
|
$cim = Get-CimInstance Win32_NetworkAdapterConfiguration -Filter "IPEnabled=True" -ErrorAction SilentlyContinue | Where-Object { $_.InterfaceIndex -eq $index } | Select-Object -First 1
|
|
[PSCustomObject]@{
|
|
name = $alias
|
|
description = $_.InterfaceDescription
|
|
mac = if ($adapter) { $adapter.MacAddress } else { "" }
|
|
status = if ($adapter) { [string]$adapter.Status } else { "" }
|
|
link_speed = if ($adapter) { [string]$adapter.LinkSpeed } else { "" }
|
|
interface_index = $index
|
|
ipv4 = @($_.IPv4Address | Select-Object -ExpandProperty IPAddress)[0]
|
|
ipv6 = @($_.IPv6Address | Select-Object -ExpandProperty IPAddress)
|
|
prefix_length = @($_.IPv4Address | Select-Object -ExpandProperty PrefixLength)[0]
|
|
gateway = @($_.IPv4DefaultGateway | Select-Object -ExpandProperty NextHop)[0]
|
|
dns = @($_.DNSServer.ServerAddresses)
|
|
dhcp_server = if ($cim) { $cim.DHCPServer } else { "" }
|
|
dhcp_lease_obtained = if ($cim) { [string]$cim.DHCPLeaseObtained } else { "" }
|
|
dhcp_lease_expires = if ($cim) { [string]$cim.DHCPLeaseExpires } else { "" }
|
|
dhcp_enabled = if ($ipif) { [string]$ipif.Dhcp -eq "Enabled" } else { $false }
|
|
}
|
|
}
|
|
$items | ConvertTo-Json -Depth 5 -Compress
|
|
"""
|
|
result = run_hidden(["powershell", "-NoProfile", "-ExecutionPolicy", "Bypass", "-Command", script], timeout=15)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(result.stdout.strip() or "PowerShell 获取网卡信息失败")
|
|
|
|
output = result.stdout.strip()
|
|
json_start = min([idx for idx in (output.find("["), output.find("{")) if idx >= 0], default=-1)
|
|
if json_start < 0:
|
|
return self._get_network_info_from_ipconfig()
|
|
|
|
data = json.loads(output[json_start:])
|
|
if isinstance(data, dict):
|
|
data = [data]
|
|
|
|
adapters = []
|
|
for item in data:
|
|
dns = item.get("dns") or []
|
|
if isinstance(dns, str):
|
|
dns = [dns]
|
|
ipv6 = item.get("ipv6") or []
|
|
if isinstance(ipv6, str):
|
|
ipv6 = [ipv6]
|
|
adapters.append(
|
|
{
|
|
"name": item.get("name") or "",
|
|
"description": item.get("description") or "",
|
|
"mac": item.get("mac") or "",
|
|
"status": item.get("status") or "",
|
|
"link_speed": item.get("link_speed") or "",
|
|
"interface_index": item.get("interface_index") or "",
|
|
"ipv4": item.get("ipv4") or "",
|
|
"ipv6": ipv6,
|
|
"prefix_length": item.get("prefix_length") or "",
|
|
"netmask": prefix_to_netmask(item.get("prefix_length")),
|
|
"gateway": item.get("gateway") or "",
|
|
"dns1": dns[0] if len(dns) > 0 else "",
|
|
"dns2": dns[1] if len(dns) > 1 else "",
|
|
"dhcp_server": item.get("dhcp_server") or "",
|
|
"dhcp_lease_obtained": item.get("dhcp_lease_obtained") or "",
|
|
"dhcp_lease_expires": item.get("dhcp_lease_expires") or "",
|
|
"dhcp_enabled": bool(item.get("dhcp_enabled")),
|
|
}
|
|
)
|
|
return sorted(adapters, key=lambda value: value["name"])
|
|
|
|
def _get_network_info_from_ipconfig(self) -> list[dict]:
|
|
result = run_hidden(["ipconfig", "/all"], timeout=15)
|
|
if result.returncode != 0 or not result.stdout.strip():
|
|
raise RuntimeError(result.stdout.strip() or "无法读取 ipconfig /all 输出")
|
|
|
|
adapters = []
|
|
current = None
|
|
last_key = None
|
|
header_pattern = re.compile(r"^(?:\S.* adapter|.+适配器)\s+(.+):$")
|
|
unknown_header_pattern = re.compile(r"^(?:Unknown adapter|未知适配器)\s+(.+):$")
|
|
value_pattern = re.compile(r"^\s*([^:]+?)\s*:\s*(.*)$")
|
|
|
|
def push_current():
|
|
if current and current.get("name"):
|
|
adapters.append(current.copy())
|
|
|
|
for raw_line in result.stdout.splitlines():
|
|
line = raw_line.rstrip()
|
|
header = header_pattern.match(line) or unknown_header_pattern.match(line)
|
|
if header:
|
|
push_current()
|
|
current = {
|
|
"name": header.group(1).strip(),
|
|
"description": "",
|
|
"mac": "",
|
|
"status": "Disconnected" if "Media disconnected" in line else "",
|
|
"link_speed": "",
|
|
"interface_index": "",
|
|
"ipv4": "",
|
|
"ipv6": [],
|
|
"prefix_length": "",
|
|
"netmask": "",
|
|
"gateway": "",
|
|
"dns1": "",
|
|
"dns2": "",
|
|
"dhcp_server": "",
|
|
"dhcp_lease_obtained": "",
|
|
"dhcp_lease_expires": "",
|
|
"dhcp_enabled": False,
|
|
}
|
|
last_key = None
|
|
continue
|
|
|
|
if current is None:
|
|
continue
|
|
|
|
if ("Media State" in line and "Media disconnected" in line) or ("媒体状态" in line and "媒体已断开连接" in line):
|
|
current["status"] = "Disconnected"
|
|
continue
|
|
|
|
match = value_pattern.match(line)
|
|
if match:
|
|
key = match.group(1).replace(".", "").strip()
|
|
value = self._clean_ipconfig_value(match.group(2))
|
|
last_key = key
|
|
self._assign_ipconfig_value(current, key, value)
|
|
continue
|
|
|
|
continuation = line.strip()
|
|
if continuation and last_key == "DNS Servers":
|
|
self._assign_ipconfig_value(current, last_key, self._clean_ipconfig_value(continuation))
|
|
|
|
push_current()
|
|
return sorted(adapters, key=lambda value: value["name"])
|
|
|
|
def _assign_ipconfig_value(self, adapter: dict, key: str, value: str) -> None:
|
|
if key == "Description":
|
|
adapter["description"] = value
|
|
elif key == "描述":
|
|
adapter["description"] = value
|
|
elif key == "Physical Address":
|
|
adapter["mac"] = value
|
|
elif key == "物理地址":
|
|
adapter["mac"] = value
|
|
elif key == "DHCP Enabled":
|
|
adapter["dhcp_enabled"] = value.lower() == "yes" or value == "是"
|
|
elif key == "DHCP 已启用":
|
|
adapter["dhcp_enabled"] = value.lower() == "yes" or value == "是"
|
|
elif key in ("IPv4 Address", "Autoconfiguration IPv4 Address"):
|
|
adapter["ipv4"] = value
|
|
if not adapter["status"]:
|
|
adapter["status"] = "Up"
|
|
elif key in ("IPv4 地址", "自动配置 IPv4 地址"):
|
|
adapter["ipv4"] = value
|
|
if not adapter["status"]:
|
|
adapter["status"] = "Up"
|
|
elif key == "Link-local IPv6 Address" or key == "IPv6 Address":
|
|
if value:
|
|
adapter["ipv6"].append(value)
|
|
elif key == "本地链接 IPv6 地址" or key == "IPv6 地址":
|
|
if value:
|
|
adapter["ipv6"].append(value)
|
|
elif key == "Subnet Mask":
|
|
adapter["netmask"] = value
|
|
elif key == "子网掩码":
|
|
adapter["netmask"] = value
|
|
elif key == "Default Gateway" and value:
|
|
adapter["gateway"] = value
|
|
elif key == "默认网关" and value:
|
|
adapter["gateway"] = value
|
|
elif key == "DNS Servers" and value:
|
|
if not adapter["dns1"]:
|
|
adapter["dns1"] = value
|
|
elif not adapter["dns2"]:
|
|
adapter["dns2"] = value
|
|
elif key == "DNS 服务器" and value:
|
|
if not adapter["dns1"]:
|
|
adapter["dns1"] = value
|
|
elif not adapter["dns2"]:
|
|
adapter["dns2"] = value
|
|
elif key == "DHCP Server":
|
|
adapter["dhcp_server"] = value
|
|
elif key == "DHCP 服务器":
|
|
adapter["dhcp_server"] = value
|
|
elif key == "Lease Obtained":
|
|
adapter["dhcp_lease_obtained"] = value
|
|
elif key == "获得租约的时间":
|
|
adapter["dhcp_lease_obtained"] = value
|
|
elif key == "Lease Expires":
|
|
adapter["dhcp_lease_expires"] = value
|
|
elif key == "租约过期的时间":
|
|
adapter["dhcp_lease_expires"] = value
|
|
|
|
def _clean_ipconfig_value(self, value: str) -> str:
|
|
return re.sub(r"\s*\((?:Preferred|首选)\)\s*$", "", value.strip())
|
|
|
|
def set_network_info(self, settings: dict) -> None:
|
|
name = settings["name"].strip()
|
|
if not name:
|
|
raise ValueError("请选择网卡")
|
|
|
|
self.output("准备应用网卡配置:\n", "muted")
|
|
for key in ("name", "dhcp_enabled", "ipv4", "netmask", "gateway", "dns1", "dns2"):
|
|
self.output(f" {key}: {settings.get(key, '')}\n", "muted")
|
|
|
|
if settings.get("dhcp_enabled"):
|
|
self._run_netsh(["interface", "ip", "set", "address", f"name={name}", "source=dhcp"])
|
|
self._run_netsh(["interface", "ip", "set", "dnsservers", f"name={name}", "source=dhcp"])
|
|
self.output("已切换为 DHCP 自动获取\n", "success")
|
|
return
|
|
|
|
ipv4 = validate_ip(settings.get("ipv4", ""))
|
|
netmask = validate_ip(settings.get("netmask", ""))
|
|
gateway = validate_ip(settings.get("gateway", ""))
|
|
self._run_netsh(
|
|
[
|
|
"interface",
|
|
"ip",
|
|
"set",
|
|
"address",
|
|
f"name={name}",
|
|
"source=static",
|
|
f"addr={ipv4}",
|
|
f"mask={netmask}",
|
|
f"gateway={gateway}",
|
|
]
|
|
)
|
|
|
|
dns1 = validate_ip(settings.get("dns1", ""), allow_empty=True)
|
|
dns2 = validate_ip(settings.get("dns2", ""), allow_empty=True)
|
|
if dns1:
|
|
self._run_netsh(["interface", "ip", "set", "dnsservers", f"name={name}", "source=static", f"address={dns1}", "index=1"])
|
|
if dns2:
|
|
self._run_netsh(["interface", "ip", "add", "dnsservers", f"name={name}", f"address={dns2}", "index=2"])
|
|
else:
|
|
self._run_netsh(["interface", "ip", "set", "dnsservers", f"name={name}", "source=dhcp"])
|
|
|
|
self.output("静态 IPv4 配置已应用\n", "success")
|
|
|
|
def load_profiles(self) -> dict:
|
|
if not os.path.exists(self.profiles_file):
|
|
return {}
|
|
with open(self.profiles_file, "r", encoding="utf-8") as file:
|
|
data = json.load(file)
|
|
return data if isinstance(data, dict) else {}
|
|
|
|
def save_profile(self, profile_name: str, settings: dict) -> None:
|
|
name = profile_name.strip()
|
|
if not name:
|
|
raise ValueError("请输入模板名称")
|
|
profiles = self.load_profiles()
|
|
profiles[name] = {
|
|
"dhcp_enabled": bool(settings.get("dhcp_enabled")),
|
|
"ipv4": settings.get("ipv4", ""),
|
|
"netmask": settings.get("netmask", ""),
|
|
"gateway": settings.get("gateway", ""),
|
|
"dns1": settings.get("dns1", ""),
|
|
"dns2": settings.get("dns2", ""),
|
|
}
|
|
self._write_profiles(profiles)
|
|
|
|
def delete_profile(self, profile_name: str) -> None:
|
|
profiles = self.load_profiles()
|
|
if profile_name in profiles:
|
|
del profiles[profile_name]
|
|
self._write_profiles(profiles)
|
|
|
|
def _write_profiles(self, profiles: dict) -> None:
|
|
with open(self.profiles_file, "w", encoding="utf-8") as file:
|
|
json.dump(profiles, file, ensure_ascii=False, indent=2)
|
|
|
|
def _run_netsh(self, args: list[str]) -> None:
|
|
command = ["netsh"] + args
|
|
result = subprocess.run(command, capture_output=True, text=True, encoding=preferred_encoding(), errors="replace")
|
|
if result.returncode != 0:
|
|
message = (result.stdout + result.stderr).strip()
|
|
raise RuntimeError(message or "netsh 命令执行失败，请确认已用管理员权限运行")
|