From 129dc412086aa7019deb6b23e9813fe4ffc5d0b3 Mon Sep 17 00:00:00 2001 From: qsc Date: Sat, 6 Jun 2026 01:57:43 +0800 Subject: [PATCH] OK --- README.md | 35 +- core/Function/common.py | 108 +++++ core/Function/network_fun.py | 386 +++++++++++++----- core/Function/ping_fun.py | 456 +++++++++++++++------ core/Function/telnet_fun.py | 770 +++++++++++++++++++++++++---------- core/Function/tracert_fun.py | 101 ++--- core/logger_config.py | 30 +- core/ui/basic_ui.py | 135 +++--- core/ui/components.py | 144 +++++++ core/ui/tab_network.py | 360 ++++++++++------ core/ui/tab_ping.py | 314 +++++++++----- core/ui/tab_telnet.py | 413 ++++++++++++++----- core/ui/tab_tracert.py | 84 ++-- core/ui/theme.py | 110 +++++ core/ui/ui_main.py | 137 ++++++- main.py | 28 +- 打包.bat | 19 +- 17 files changed, 2582 insertions(+), 1048 deletions(-) create mode 100644 core/Function/common.py create mode 100644 core/ui/components.py create mode 100644 core/ui/theme.py diff --git a/README.md b/README.md index e378dfc..cda2349 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,35 @@ # Network-tools -常用的网络调试工具 + +一个基于 Tkinter 的 Windows 网络调试工具，重构版命名为 **NetPilot**。 + +## 功能 + +- 网卡配置：读取网卡、IPv4/IPv6、网关、DNS、DHCP、链路速率、接口索引、DHCP 服务器与租约时间。 +- 配置模板：保存常用网卡配置、套用模板到表单、删除模板。 +- Ping 探测：支持持续/指定次数、间隔、超时、包大小、TTL、禁止分片、本地源 IP 下拉、CIDR/范围/列表批量探活、目标导入、CSV 导出、实时丢包率/平均延迟/抖动/质量统计。 +- 端口扫描：支持单端口测试、端口列表/范围、常用端口预设、批量主机/CIDR/IP 段扫描、并发/超时控制、只显示开放端口、服务名识别、可选 Banner 探测、开放端口复制与 CSV 导出。 +- 路由追踪：支持设置最大跳数和单跳超时时间。 + +## 运行 + +```powershell +python main.py +``` + +修改网卡配置需要以管理员权限运行程序。 + +## 打包 + +直接执行： + +```powershell +.\打包.bat +``` + +打包输出： + +```text +dist/NetworkTool.exe +``` + +配置模板运行时会保存到程序同级目录的 `network_profiles.json`。 diff --git a/core/Function/common.py b/core/Function/common.py new file mode 100644 index 0000000..7d4d06f --- /dev/null +++ b/core/Function/common.py @@ -0,0 +1,108 @@ +import ipaddress +import locale +import re +import subprocess +import sys +from typing import Iterable, List + + +CREATE_NO_WINDOW = subprocess.CREATE_NO_WINDOW if sys.platform.startswith("win") else 0 + + +def preferred_encoding() -> str: + return locale.getpreferredencoding(False) or "utf-8" + + +def run_hidden(command: List[str], timeout=None) -> subprocess.CompletedProcess: + return subprocess.run( + command, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + encoding=preferred_encoding(), + errors="replace", + timeout=timeout, + creationflags=CREATE_NO_WINDOW, + ) + + +def popen_hidden(command: List[str]) -> subprocess.Popen: + return subprocess.Popen( + command, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + encoding=preferred_encoding(), + errors="replace", + creationflags=CREATE_NO_WINDOW, + ) + + +def validate_host(value: str) -> str: + host = value.strip() + if not host: + raise ValueError("请输入目标地址或域名") + return host + + +def validate_ip(value: str, allow_empty: bool = False) -> str: + ip = value.strip() + if allow_empty and not ip: + return "" + try: + ipaddress.ip_address(ip) + except ValueError as exc: + raise ValueError(f"IP 地址无效: {value}") from exc + return ip + + +def validate_port(value, label: str = "端口") -> int: + try: + port = int(str(value).strip()) + except ValueError as exc: + raise ValueError(f"{label}必须是整数") from exc + if port < 1 or port > 65535: + raise ValueError(f"{label}必须在 1-65535 之间") + return port + + +def parse_ports(text: str) -> List[int]: + ports = set() + for part in re.split(r"[,，;\s]+", text.strip()): + part = part.strip() + if not part: + continue + if "-" in part: + start_raw, end_raw = part.split("-", 1) + start = validate_port(start_raw, "起始端口") + end = validate_port(end_raw, "结束端口") + if start > end: + raise ValueError("端口范围的起始值不能大于结束值") + ports.update(range(start, end + 1)) + else: + ports.add(validate_port(part)) + if not ports: + raise ValueError("请输入至少一个端口") + return sorted(ports) + + +def prefix_to_netmask(prefix_length) -> str: + if prefix_length in ("", None): + return "" + try: + prefix = int(prefix_length) + return str(ipaddress.IPv4Network(f"0.0.0.0/{prefix}").netmask) + except Exception: + return "" + + +def iter_hosts(prefix: str, start: int, end: int) -> Iterable[str]: + if not prefix.endswith("."): + raise ValueError("网段前缀必须以点号结尾，例如 192.168.1.") + for value in (start, end): + if value < 1 or value > 254: + raise ValueError("主机号范围必须在 1-254 之间") + if start > end: + raise ValueError("起始主机号不能大于结束主机号") + for host_id in range(start, end + 1): + yield validate_ip(f"{prefix}{host_id}") diff --git a/core/Function/network_fun.py b/core/Function/network_fun.py index c6425c4..7a77ec6 100644 --- a/core/Function/network_fun.py +++ b/core/Function/network_fun.py @@ -1,129 +1,303 @@ -import subprocess -import locale +import json +import os import re -import tkinter as tk -from tkinter import scrolledtext, messagebox +import subprocess +import sys +from typing import Callable, Optional -import logging +from core.Function.common import preferred_encoding, prefix_to_netmask, run_hidden, validate_ip + + +OutputCallback = Callable[[str, Optional[str]], None] class NetworkManager: - """获取本地网卡详细信息（含 DNS、DHCP）""" - def __init__(self, result_box: scrolledtext.ScrolledText): - self.result_box = result_box + def __init__(self, output: OutputCallback): + self.output = output + self.profiles_file = os.path.join(self._base_dir(), "network_profiles.json") - def get_network_info(self): - # 自动获取系统编码（例如 'cp936' 中文Windows） - system_encoding = locale.getpreferredencoding(False) - - # 调用 ipconfig /all - result = subprocess.run( - "ipconfig /all", - shell=True, - capture_output=True, - text=True, - encoding=system_encoding, # 自动根据系统语言选择 - errors="ignore" # 忽略解码错误 - ) + 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__)))) - output = result.stdout - if not output: - raise RuntimeError("无法获取 ipconfig 输出，请检查系统命令执行权限。") + 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 获取网卡信息失败") - # 按块拆分 - import re - adapter_blocks = re.split(r"\r?\n(?=\S.*?:)", output) + 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 block in adapter_blocks: - if not re.search(r"适配器", block): + 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 - info = {} - match = re.match(r"(.+?):", block) + 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: - info["name"] = match.group(1).strip() + 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 - match = re.search(r"描述.*?:\s*(.+)", block) - if match: - info["description"] = match.group(1).strip() + continuation = line.strip() + if continuation and last_key == "DNS Servers": + self._assign_ipconfig_value(current, last_key, self._clean_ipconfig_value(continuation)) - match = re.search(r"物理地址.*?:\s*(.+)", block) - if match: - info["mac"] = match.group(1).strip() + push_current() + return sorted(adapters, key=lambda value: value["name"]) - match = re.search(r"DHCP 已启用.*?:\s*(.+)", block) - if match: - value = match.group(1).strip() - info["dhcp_enabled"] = "是" if value == "是" else "否" + 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 - match = re.search(r"IPv4 地址.*?:\s*([0-9.]+)", block) - if match: - info["ipv4"] = match.group(1) + def _clean_ipconfig_value(self, value: str) -> str: + return re.sub(r"\s*\((?:Preferred|首选)\)\s*$", "", value.strip()) - match = re.search(r"子网掩码.*?:\s*([0-9.]+)", block) - if match: - info["netmask"] = match.group(1) + def set_network_info(self, settings: dict) -> None: + name = settings["name"].strip() + if not name: + raise ValueError("请选择网卡") - gateway = "" - # 先找出 "默认网关" 所在行及后续可能的下一行 - gw_match = re.search(r"默认网关[.\s:]*([^\r\n]*)\r?\n(?:\s*([^\r\n]+))?", block) - if gw_match: - # 合并两行文本 - gw_text = " ".join(gw_match.groups(default="")) - # 提取 IPv4 地址 - ipv4_match = re.search(r"\b\d{1,3}(?:\.\d{1,3}){3}\b", gw_text) - if ipv4_match: - gateway = ipv4_match.group(0) - info["gateway"] = gateway + self.output("准备应用网卡配置:\n", "muted") + for key in ("name", "dhcp_enabled", "ipv4", "netmask", "gateway", "dns1", "dns2"): + self.output(f" {key}: {settings.get(key, '')}\n", "muted") - # --- DNS 服务器提取（支持多行 + IPv4优先）--- - dns1 = dns2 = "" - # 找出 “DNS 服务器” 开始的位置 - dns_match = re.search(r"DNS 服务器[.\s:]*([^\r\n]*)((?:\r?\n\s+[^\r\n]+)*)", block) - if dns_match: - # 合并所有行 - dns_text = dns_match.group(1) + dns_match.group(2) - # 提取所有 IPv4 地址（优先），如果没有，再取 IPv6 - dns_ipv4 = re.findall(r"\b\d{1,3}(?:\.\d{1,3}){3}\b", dns_text) - if dns_ipv4: - dns1 = dns_ipv4[0] - if len(dns_ipv4) > 1: - dns2 = dns_ipv4[1] - else: - # 没有 IPv4，则尝试 IPv6 - dns_ipv6 = re.findall(r"[a-fA-F0-9:]+(?:%[0-9]+)?", dns_text) - if dns_ipv6: - dns1 = dns_ipv6[0] - if len(dns_ipv6) > 1: - dns2 = dns_ipv6[1] + 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 - info["dns1"] = dns1 - info["dns2"] = dns2 + 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}", + ] + ) - adapters.append(info) - return adapters - - def set_network_info(self, settings): - """设置指定网卡的网络配置""" - # 自动获取系统编码（例如 'cp936' 中文Windows） - system_encoding = locale.getpreferredencoding(False) - try: - self.result_box.insert(tk.END, f"尝试将网卡配置修改为：\n") - for key, value in settings.items(): - self.result_box.insert(tk.END, f" {key}: {value}\n") - if settings['dhcp_enabled']: - # 启用 DHCP 自动获取 IP 地址 - subprocess.run( - f'netsh interface ip set address name="{settings['name']}" source=dhcp', - shell=True, check=True - ) - else: - subprocess.run( - f'netsh interface ip set address name="{settings['name']}" source=static addr={settings['ipv4']} mask={settings['netmask']} gateway={settings['gateway']}', - shell=True, check=True) - self.result_box.insert(tk.END, f"网络配置设置完成\n") + 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"]) - except subprocess.CalledProcessError as e: - self.result_box.insert(tk.END, f"设置{settings['name']}网络配置失败: {e}\n") - self.result_box.see(tk.END) \ No newline at end of file + 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 命令执行失败，请确认已用管理员权限运行") diff --git a/core/Function/ping_fun.py b/core/Function/ping_fun.py index 865bf77..9dcf54e 100644 --- a/core/Function/ping_fun.py +++ b/core/Function/ping_fun.py @@ -1,157 +1,345 @@ -import tkinter as tk -from tkinter import scrolledtext, messagebox -import subprocess import concurrent.futures -import threading +import csv +import ipaddress import re -import sys +import threading +import time +from dataclasses import dataclass, field +from typing import Callable, Optional + +from core.Function.common import run_hidden, validate_host, validate_ip + + +OutputCallback = Callable[[str, Optional[str]], None] +DoneCallback = Callable[[], None] +StatusCallback = Callable[[dict], None] + + +@dataclass +class PingStats: + sent: int = 0 + received: int = 0 + rtts: list[float] = field(default_factory=list) + current_loss_streak: int = 0 + max_loss_streak: int = 0 + last_success_at: str = "" + + @property + def lost(self) -> int: + return max(self.sent - self.received, 0) + + @property + def loss_rate(self) -> float: + return self.lost / self.sent * 100 if self.sent else 0.0 + + @property + def avg_rtt(self) -> float: + return sum(self.rtts) / len(self.rtts) if self.rtts else 0.0 + + @property + def jitter(self) -> float: + if len(self.rtts) < 2: + return 0.0 + deltas = [abs(self.rtts[index] - self.rtts[index - 1]) for index in range(1, len(self.rtts))] + return sum(deltas) / len(deltas) + + @property + def quality(self) -> str: + if not self.sent: + return "等待" + if self.loss_rate >= 30 or self.max_loss_streak >= 3: + return "丢包严重" + if self.loss_rate > 0 or self.jitter >= 50: + return "波动" + return "稳定" + + def snapshot(self) -> dict: + return { + "sent": self.sent, + "received": self.received, + "lost": self.lost, + "loss_rate": self.loss_rate, + "min_rtt": min(self.rtts) if self.rtts else 0.0, + "max_rtt": max(self.rtts) if self.rtts else 0.0, + "avg_rtt": self.avg_rtt, + "jitter": self.jitter, + "max_loss_streak": self.max_loss_streak, + "last_success_at": self.last_success_at, + "quality": self.quality, + } + class PingFun: - def __init__(self, result_box: scrolledtext.ScrolledText): - self.result_box = result_box + def __init__( + self, + output: OutputCallback, + done: Optional[DoneCallback] = None, + status: Optional[StatusCallback] = None, + ): + self.output = output + self.done = done or (lambda: None) + self.status = status or (lambda _stats: None) + self.stop_event = threading.Event() + self.worker = None + self.batch_worker = None + self.stats = PingStats() + self.last_batch_results = [] - # Ping 状态和统计 - self.process = None - self.stop_flag = False - self.sent = 0 - self.received = 0 - self.rtts = [] + def start_ping(self, host: str, local_ip: str = "", options: Optional[dict] = None) -> None: + host = validate_host(host) + local_ip = validate_ip(local_ip, allow_empty=True) + options = self.normalize_options(options) - def strat_ping(self, host, local_ip=None, callback=None): - """开始 ping""" - self.callback = callback - self.result_box.delete('1.0', tk.END) - self.sent = 0 - self.received = 0 - self.rtts.clear() - self.stop_flag = False + if self.is_running() or self.is_batch_running(): + raise RuntimeError("Ping 正在运行，请先停止当前任务") - command = ['ping', host, '-t'] - if local_ip: - command += ['-S', local_ip] + self.stop_event.clear() + self.stats = PingStats() + self.status(self.stats.snapshot() | {"state": "运行中"}) - threading.Thread(target=self.ping, args=(command,), daemon=True).start() + self.output(f"开始 Ping: {host}\n", "muted") + self.output(self.describe_options(options, local_ip), "muted") + self.worker = threading.Thread(target=self._run_ping_loop, args=(host, local_ip, options), daemon=True) + self.worker.start() - def stop_ping(self): - """手动停止 ping""" - if self.process: - self.process.terminate() - self.result_box.insert(tk.END, "\nPing 已手动停止。\n") - self.result_box.see(tk.END) - self.process = None - self.show_statistics() - - def ping(self, command): - """执行 ping 命令并处理输出""" - rtt_pattern = re.compile(r'时间[=<](\d+)ms', re.IGNORECASE) + def _run_ping_loop(self, host: str, local_ip: str, options: dict) -> None: + max_count = None if options["mode"] == "持续" else options["count"] try: - # 👇关键：隐藏 CMD 窗口 - creationflags = subprocess.CREATE_NO_WINDOW if sys.platform.startswith("win") else 0 - - self.process = subprocess.Popen( - command, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - universal_newlines=True, - creationflags=creationflags - ) - for line in iter(self.process.stdout.readline, ''): - if self.stop_flag: - break - if line: - self.result_box.insert(tk.END, line) - self.result_box.see(tk.END) - self.sent += 1 - match = rtt_pattern.search(line) - if match: - self.received += 1 - self.rtts.append(float(match.group(1))) - except Exception as e: - self.result_box.insert(tk.END, f"Ping 失败: {e}\n") + while not self.stop_event.is_set() and (max_count is None or self.stats.sent < max_count): + started = time.perf_counter() + result = self.ping_once(host, local_ip, options) + self._record_single_result(result) + elapsed_ms = (time.perf_counter() - started) * 1000 + wait_ms = max(options["interval_ms"] - elapsed_ms, 0) + self.stop_event.wait(wait_ms / 1000) finally: - self.process = None - if self.callback: - self.callback() + self._write_statistics("Ping 统计") + self.status(self.stats.snapshot() | {"state": "已停止" if self.stop_event.is_set() else "已完成"}) + self.done() - def show_statistics(self): - if self.sent == 0: - return - loss = (self.sent - self.received) / self.sent * 100 - stats = f"\n==== Ping 统计 ====\n" \ - f"发送: {self.sent}，接收: {self.received}，丢包率: {loss:.2f}%\n" - if self.rtts: - stats += f"最小延迟: {min(self.rtts)} ms，最大延迟: {max(self.rtts)} ms，平均延迟: {sum(self.rtts)/len(self.rtts):.2f} ms\n" - self.result_box.insert(tk.END, stats) - self.result_box.see(tk.END) + def _record_single_result(self, result: dict) -> None: + self.stats.sent += 1 + if result["ok"]: + self.stats.received += 1 + self.stats.current_loss_streak = 0 + self.stats.rtts.append(result["rtt"]) + self.stats.last_success_at = time.strftime("%H:%M:%S") + self.output(f"[{self.stats.sent}] {result['host']} 通 {result['rtt']:.1f} ms\n", "success") + else: + self.stats.current_loss_streak += 1 + self.stats.max_loss_streak = max(self.stats.max_loss_streak, self.stats.current_loss_streak) + self.output(f"[{self.stats.sent}] {result['host']} {result['message']}\n", "warning") + self.status(self.stats.snapshot() | {"state": "运行中"}) - # ================= 批量 Ping ================= - def start_batch_ping(self, net_prefix, start, end, local_ip=None, callback=None): - self.callback = callback - self.stop_flag = False - self.result_box.delete('1.0', tk.END) - self.result_box.insert(tk.END, f"开始并发 Ping：{net_prefix}{start} - {net_prefix}{end}\n\n") + def stop_ping(self) -> None: + self.stop_event.set() + self.output("\n已请求停止 Ping\n", "warning") - self.batch_thread = threading.Thread( - target=self._concurrent_batch_ping, args=(net_prefix, start, end, local_ip), daemon=True) - self.batch_thread.start() + def start_batch_ping(self, target_text: str, local_ip: str = "", options: Optional[dict] = None) -> None: + local_ip = validate_ip(local_ip, allow_empty=True) + hosts = parse_ping_targets(target_text) + options = self.normalize_options(options) - def _ping_one_ip(self, ip, local_ip=None): - """Ping 单个 IP 地址并返回结果""" - if self.stop_flag: - return None + if self.is_running() or self.is_batch_running(): + raise RuntimeError("Ping 正在运行，请先停止当前任务") - command = ['ping', ip, '-n', '1', '-w', '2000'] - if local_ip: - command += ['-S', local_ip] + self.stop_event.clear() + self.last_batch_results = [] + self.status({"state": "批量运行中", "sent": 0, "received": 0, "lost": 0, "loss_rate": 0, "avg_rtt": 0, "jitter": 0, "quality": "等待"}) + self.output(f"开始批量 Ping: 共 {len(hosts)} 个目标\n", "muted") + self.output(self.describe_options(options, local_ip), "muted") + self.batch_worker = threading.Thread(target=self._run_batch_ping, args=(hosts, local_ip, options), daemon=True) + self.batch_worker.start() + + def _run_batch_ping(self, hosts: list[str], local_ip: str, options: dict) -> None: + results = [] + done = 0 + workers = min(options["workers"], max(1, len(hosts))) + + def task(host: str) -> dict: + if self.stop_event.is_set(): + return {"host": host, "ok": False, "rtt": 0.0, "message": "已取消"} + return self.ping_once(host, local_ip, options) try: - # 👇 同样隐藏 CMD 窗口 - creationflags = subprocess.CREATE_NO_WINDOW if sys.platform.startswith("win") else 0 - - result = subprocess.run( - command, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - timeout=2, - creationflags=creationflags # ✅ 加上 - ) - - if "TTL=" in result.stdout.upper(): - return f"{ip} ✅ 通\n" + with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor: + future_to_host = {executor.submit(task, host): host for host in hosts} + for future in concurrent.futures.as_completed(future_to_host): + if self.stop_event.is_set(): + for item in future_to_host: + item.cancel() + break + result = future.result() + done += 1 + results.append(result) + tag = "success" if result["ok"] else None + message = f"{result['rtt']:.1f} ms" if result["ok"] else result["message"] + self.output(f"[{done}/{len(hosts)}] {result['host']:<15} {message}\n", tag) + self.status(self._batch_snapshot(results, len(hosts)) | {"state": "批量运行中"}) + finally: + ordered = sorted(results, key=lambda item: ipaddress.ip_address(item["host"])) + self.last_batch_results = ordered + if self.stop_event.is_set(): + self.output("\n批量 Ping 已停止\n", "warning") else: - return f"{ip} ❌ 不通\n" - except subprocess.TimeoutExpired: - return f"{ip} ⚠️ 超时\n" - except Exception as e: - return f"{ip} 错误: {e}\n" + self._write_batch_summary(ordered, len(hosts)) + self.status(self._batch_snapshot(ordered, len(hosts)) | {"state": "已停止" if self.stop_event.is_set() else "已完成"}) + self.done() - def _concurrent_batch_ping(self, net_prefix, start, end, local_ip=None): - '''并发批量 Ping''' - ip_list = [f"{net_prefix}{i}" for i in range(start, end + 1)] - max_workers = min(50, len(ip_list)) - with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: - future_to_ip = {executor.submit(self._ping_one_ip, ip, local_ip): ip for ip in ip_list} - for future in concurrent.futures.as_completed(future_to_ip): - if self.stop_flag: - break - result = future.result() - if result: - self.result_box.insert(tk.END, result) - self.result_box.see(tk.END) + def stop_batch_ping(self) -> None: + self.stop_event.set() + self.output("\n正在停止批量 Ping...\n", "warning") - if not self.stop_flag: - self.result_box.insert(tk.END, "\n并发批量 Ping 完成。\n") - else: - self.result_box.insert(tk.END, "\n批量 Ping 已停止。\n") - self.result_box.see(tk.END) - if self.callback: - self.callback() + def ping_once(self, host: str, local_ip: str, options: dict) -> dict: + command = ["ping", host, "-n", "1", "-w", str(options["timeout_ms"])] + if local_ip: + command += ["-S", local_ip] + if options["size"] > 0: + command += ["-l", str(options["size"])] + if options["ttl"] > 0: + command += ["-i", str(options["ttl"])] + if options["dont_fragment"]: + command.append("-f") - def stop_batch_ping(self): - if not self.batch_thread or not self.batch_thread.is_alive(): - messagebox.showinfo("提示", "当前没有正在运行的批量 Ping。") + try: + result = run_hidden(command, timeout=max(2, options["timeout_ms"] / 1000 + 2)) + return parse_ping_output(host, result.stdout) + except Exception as exc: + return {"host": host, "ok": False, "rtt": 0.0, "message": str(exc)} + + def normalize_options(self, options: Optional[dict]) -> dict: + options = options or {} + return { + "mode": options.get("mode", "持续"), + "count": clamp_int(options.get("count", 4), 1, 100000, "次数"), + "interval_ms": clamp_int(options.get("interval_ms", 1000), 100, 60000, "间隔"), + "timeout_ms": clamp_int(options.get("timeout_ms", 1200), 100, 60000, "超时"), + "size": clamp_int(options.get("size", 32), 0, 65500, "包大小"), + "ttl": clamp_int(options.get("ttl", 0), 0, 255, "TTL"), + "dont_fragment": bool(options.get("dont_fragment", False)), + "workers": clamp_int(options.get("workers", 64), 1, 256, "并发数"), + } + + def describe_options(self, options: dict, local_ip: str) -> str: + mode = "持续" if options["mode"] == "持续" else f"{options['count']} 次" + source = local_ip or "默认路由" + df = "是" if options["dont_fragment"] else "否" + ttl = options["ttl"] if options["ttl"] else "默认" + return ( + f"模式: {mode} 源地址: {source} 间隔: {options['interval_ms']}ms " + f"超时: {options['timeout_ms']}ms 包大小: {options['size']} bytes TTL: {ttl} 禁止分片: {df}\n\n" + ) + + def _write_statistics(self, title: str) -> None: + snapshot = self.stats.snapshot() + if not snapshot["sent"]: return - self.stop_flag = True - self.result_box.insert(tk.END, "\n正在尝试停止批量 Ping...\n") - self.result_box.see(tk.END) + lines = [ + f"\n==== {title} ====\n", + f"发送: {snapshot['sent']} 接收: {snapshot['received']} 丢失: {snapshot['lost']} 丢包率: {snapshot['loss_rate']:.1f}%\n", + f"延迟: 最小 {snapshot['min_rtt']:.1f} ms 最大 {snapshot['max_rtt']:.1f} ms 平均 {snapshot['avg_rtt']:.1f} ms 抖动 {snapshot['jitter']:.1f} ms\n", + f"最大连续丢包: {snapshot['max_loss_streak']} 最后成功: {snapshot['last_success_at'] or '-'} 状态: {snapshot['quality']}\n", + ] + self.output("".join(lines), "success" if snapshot["loss_rate"] == 0 else "warning") + + def _write_batch_summary(self, results: list[dict], total: int) -> None: + online = [item for item in results if item["ok"]] + offline = [item for item in results if not item["ok"] and item["message"] != "超时"] + timeout = [item for item in results if item["message"] == "超时"] + snapshot = self._batch_snapshot(results, total) + + self.output("\n==== 批量 Ping 统计 ====\n", "muted") + self.output(f"总数: {total} 在线: {len(online)} 离线: {len(offline)} 超时: {len(timeout)} 成功率: {100 - snapshot['loss_rate']:.1f}%\n", "success") + if online: + self.output("在线 IP: " + ", ".join(item["host"] for item in online) + "\n", "success") + if timeout: + self.output("超时 IP: " + ", ".join(item["host"] for item in timeout) + "\n", "warning") + if offline: + self.output("离线 IP: " + ", ".join(item["host"] for item in offline) + "\n", "warning") + + def _batch_snapshot(self, results: list[dict], total: int) -> dict: + received = len([item for item in results if item["ok"]]) + rtts = [item["rtt"] for item in results if item["ok"]] + sent = len(results) + loss_rate = (sent - received) / sent * 100 if sent else 0.0 + avg = sum(rtts) / len(rtts) if rtts else 0.0 + jitter = 0.0 + if len(rtts) > 1: + deltas = [abs(rtts[index] - rtts[index - 1]) for index in range(1, len(rtts))] + jitter = sum(deltas) / len(deltas) + return { + "sent": sent, + "received": received, + "lost": max(sent - received, 0), + "loss_rate": loss_rate, + "avg_rtt": avg, + "jitter": jitter, + "quality": "稳定" if loss_rate == 0 else "波动" if loss_rate < 30 else "丢包严重", + } + + def export_batch_results(self, path: str) -> None: + if not self.last_batch_results: + raise RuntimeError("还没有可导出的批量 Ping 结果") + with open(path, "w", newline="", encoding="utf-8-sig") as file: + writer = csv.DictWriter(file, fieldnames=["host", "ok", "rtt", "message"]) + writer.writeheader() + writer.writerows(self.last_batch_results) + + def is_running(self) -> bool: + return bool(self.worker and self.worker.is_alive()) + + def is_batch_running(self) -> bool: + return bool(self.batch_worker and self.batch_worker.is_alive()) + + +def parse_ping_output(host: str, output: str) -> dict: + ttl_found = re.search(r"\bTTL=", output, re.IGNORECASE) + rtt_match = re.search(r"(?:time|时间)[=<]?\s*(\d+(?:\.\d+)?)\s*ms", output, re.IGNORECASE) + if ttl_found: + rtt = float(rtt_match.group(1)) if rtt_match else 0.0 + return {"host": host, "ok": True, "rtt": rtt, "message": "在线"} + if re.search(r"请求超时|timed out|timeout", output, re.IGNORECASE): + return {"host": host, "ok": False, "rtt": 0.0, "message": "超时"} + if re.search(r"无法访问|unreachable|could not find host|找不到主机", output, re.IGNORECASE): + return {"host": host, "ok": False, "rtt": 0.0, "message": "不可达"} + return {"host": host, "ok": False, "rtt": 0.0, "message": "无响应"} + + +def parse_ping_targets(text: str) -> list[str]: + raw = text.strip() + if not raw: + raise ValueError("请输入批量 Ping 目标") + + targets = [] + for part in re.split(r"[,，\s]+", raw): + item = part.strip() + if not item: + continue + if "/" in item: + network = ipaddress.ip_network(item, strict=False) + targets.extend(str(ip) for ip in network.hosts()) + elif re.match(r"^\d{1,3}(?:\.\d{1,3}){3}-\d{1,3}$", item): + prefix, end_text = item.rsplit(".", 1) + start_text, end_host = end_text.split("-", 1) + start = int(start_text) + end = int(end_host) + if start > end: + raise ValueError("IP 范围起始值不能大于结束值") + targets.extend(str(ipaddress.ip_address(f"{prefix}.{index}")) for index in range(start, end + 1)) + else: + targets.append(str(ipaddress.ip_address(item))) + + unique = list(dict.fromkeys(targets)) + if not unique: + raise ValueError("没有解析到有效目标") + return unique + + +def clamp_int(value, min_value: int, max_value: int, label: str) -> int: + try: + number = int(str(value).strip()) + except ValueError as exc: + raise ValueError(f"{label}必须是整数") from exc + if number < min_value or number > max_value: + raise ValueError(f"{label}必须在 {min_value}-{max_value} 之间") + return number diff --git a/core/Function/telnet_fun.py b/core/Function/telnet_fun.py index 837bf27..4759b85 100644 --- a/core/Function/telnet_fun.py +++ b/core/Function/telnet_fun.py @@ -1,251 +1,581 @@ +import concurrent.futures +import csv +import ipaddress +import re import socket import threading -import concurrent.futures -import tkinter as tk -from tkinter import scrolledtext, messagebox -from typing import Iterable, Optional, List +import time +from dataclasses import asdict, dataclass, field +from typing import Callable, Iterable, Optional + +from core.Function.common import parse_ports, validate_host, validate_port + + +OutputCallback = Callable[[str, Optional[str]], None] +DoneCallback = Callable[[], None] +StatusCallback = Callable[[dict], None] +ResultCallback = Callable[[dict], None] + + +STATUS_TEXT = { + "open": "开放", + "closed": "关闭", + "timeout": "超时", + "unreachable": "不可达", + "dns_error": "解析失败", + "cancelled": "已取消", + "error": "错误", +} + +COMMON_SERVICES = { + 20: "FTP-DATA", + 21: "FTP", + 22: "SSH", + 23: "TELNET", + 25: "SMTP", + 53: "DNS", + 67: "DHCP", + 68: "DHCP", + 80: "HTTP", + 110: "POP3", + 123: "NTP", + 135: "MSRPC", + 137: "NETBIOS", + 138: "NETBIOS", + 139: "NETBIOS", + 143: "IMAP", + 389: "LDAP", + 443: "HTTPS", + 445: "SMB", + 465: "SMTPS", + 587: "SMTP", + 636: "LDAPS", + 993: "IMAPS", + 995: "POP3S", + 1433: "MSSQL", + 1521: "ORACLE", + 3306: "MYSQL", + 3389: "RDP", + 5432: "POSTGRES", + 5900: "VNC", + 5985: "WINRM", + 5986: "WINRM-SSL", + 6379: "REDIS", + 8000: "HTTP-ALT", + 8080: "HTTP-ALT", + 8443: "HTTPS-ALT", + 9200: "ELASTIC", + 27017: "MONGODB", +} + +HTTP_BANNER_PORTS = {80, 8000, 8008, 8080, 8081, 8888, 9000} + + +@dataclass +class ScanOptions: + timeout_ms: int = 800 + workers: int = 128 + show_closed: bool = False + banner_probe: bool = False + + +@dataclass +class PortResult: + host: str + resolved_ip: str + ip_version: str + port: int + service: str + status: str + status_text: str + latency_ms: float = 0.0 + banner: str = "" + error: str = "" + checked_at: str = field(default_factory=lambda: time.strftime("%Y-%m-%d %H:%M:%S")) + + def to_dict(self) -> dict: + data = asdict(self) + data["latency_ms"] = round(self.latency_ms, 1) + return data + + +@dataclass +class ScanStats: + total: int = 0 + scanned: int = 0 + open_count: int = 0 + closed_count: int = 0 + timeout_count: int = 0 + unreachable_count: int = 0 + dns_error_count: int = 0 + error_count: int = 0 + cancelled_count: int = 0 + started_at: float = field(default_factory=time.perf_counter) + + def record(self, status: str, count: int = 1) -> None: + self.scanned += count + if status == "open": + self.open_count += count + elif status == "closed": + self.closed_count += count + elif status == "timeout": + self.timeout_count += count + elif status == "unreachable": + self.unreachable_count += count + elif status == "dns_error": + self.dns_error_count += count + elif status == "cancelled": + self.cancelled_count += count + else: + self.error_count += count + + def snapshot(self, state: str) -> dict: + elapsed = time.perf_counter() - self.started_at + progress = self.scanned / self.total * 100 if self.total else 0.0 + return { + "state": state, + "total": self.total, + "scanned": self.scanned, + "open": self.open_count, + "closed": self.closed_count, + "timeout": self.timeout_count, + "unreachable": self.unreachable_count, + "dns_error": self.dns_error_count, + "error": self.error_count, + "cancelled": self.cancelled_count, + "progress": progress, + "elapsed": elapsed, + } class PortScanner: - """ - PortScanner: 在 Tkinter 的 ScrolledText 中显示端口检测结果的工具类。 + def __init__( + self, + output: OutputCallback, + done: Optional[DoneCallback] = None, + status: Optional[StatusCallback] = None, + result: Optional[ResultCallback] = None, + ): + self.output = output + self.done = done or (lambda: None) + self.status = status or (lambda _stats: None) + self.result = result or (lambda _result: None) + self.stop_event = threading.Event() + self.scan_thread = None + self.executor = None + self.last_results: list[dict] = [] - 构造： - scanner = PortScanner(result_box) + def test_connect(self, host: str, port: int, timeout: float = 1.0, options: Optional[dict] = None) -> None: + options = self.normalize_options(options or {"timeout_ms": int(timeout * 1000), "workers": 1, "show_closed": True}) + host = validate_host(host) + port = validate_port(port) + self._start_scan([host], [port], options, "单端口测试") - 方法： - test_connect(ip, port, timeout=1.0) - - 直接测试单个 ip:port 是否可连通（立即返回 bool，并在结果框显示一行结果） + def start_text_scan(self, host: str, ports_text: str, timeout: float = 0.8, options: Optional[dict] = None) -> None: + options = self.normalize_options(options or {"timeout_ms": int(timeout * 1000)}) + self.start_list_scan(host, parse_ports(ports_text), timeout=timeout, options=options) - start_range_scan(ip, start_port, end_port, timeout=1.0, max_workers=100) - - 并发扫描端口范围 [start_port, end_port]（包含端口边界） - - 扫描结果会实时写入 result_box + def start_range_scan(self, host: str, start_port: int, end_port: int, timeout: float = 0.8, options: Optional[dict] = None) -> None: + start_port = validate_port(start_port, "起始端口") + end_port = validate_port(end_port, "结束端口") + if start_port > end_port: + raise ValueError("起始端口不能大于结束端口") + options = self.normalize_options(options or {"timeout_ms": int(timeout * 1000)}) + self.start_list_scan(host, range(start_port, end_port + 1), timeout=timeout, options=options) - start_list_scan(ip, ports: Iterable[int], timeout=1.0, max_workers=100) - - 并发扫描给定端口列表 + def start_list_scan( + self, + host: str, + ports: Iterable[int], + timeout: float = 0.8, + options: Optional[dict] = None, + ) -> None: + host = validate_host(host) + ports = normalize_ports(ports) + options = self.normalize_options(options or {"timeout_ms": int(timeout * 1000)}) + self._start_scan([host], ports, options, "端口扫描") - stop_scan() - - 尝试中止正在进行的扫描（设置停止标志，后续任务检测到后会停止提交或返回） + def start_scan(self, host: str, ports_text: str, options: Optional[dict] = None) -> None: + host = validate_host(host) + ports = parse_ports(ports_text) + self._start_scan([host], ports, self.normalize_options(options), "端口扫描") - 注意： - - 使用 TCP 连接测试（socket.connect），适合服务端口检测。 - - GUI 写入通过 self._append_text(...) 调度到主线程，保证线程安全。 - """ + def start_batch_scan(self, hosts_text: str, ports_text: str, options: Optional[dict] = None) -> None: + hosts = parse_scan_hosts(hosts_text) + ports = parse_ports(ports_text) + self._start_scan(hosts, ports, self.normalize_options(options), "批量主机扫描") - def __init__(self, result_box: scrolledtext.ScrolledText): - self.result_box = result_box + def _start_scan(self, hosts: list[str], ports: list[int], options: ScanOptions, title: str) -> None: + if self.is_scanning(): + raise RuntimeError("端口扫描正在运行，请先停止当前任务") + if not ports: + raise ValueError("请输入至少一个端口") - # 扫描控制状态 - self._stop_flag = False # 外部调用 stop_scan() 会把此标志设为 True - self._scan_thread: Optional[threading.Thread] = None - self._executor: Optional[concurrent.futures.ThreadPoolExecutor] = None + self.stop_event.clear() + self.last_results = [] + total = len(hosts) * len(ports) + self.output(f"开始{title}: {len(hosts)} 个目标，{len(ports)} 个端口，共 {total} 次连接\n", "muted") + self.output(self.describe_options(options), "muted") + self.status(ScanStats(total=total).snapshot("运行中")) + self.scan_thread = threading.Thread(target=self._scan, args=(hosts, ports, options, title), daemon=True) + self.scan_thread.start() - # 用于统计（可选） - self._total = 0 - self._done = 0 - self._open_ports: List[int] = [] - - # ------------------------- - # 辅助方法：线程安全地向结果框写文本 - # ------------------------- - def _append_text(self, text: str): - """ - 把文本插入到 result_box。因为可能从子线程调用，所以用 after 调度到主线程执行。 - """ - try: - # schedule on main thread immediately - self.result_box.after(0, lambda: (self.result_box.insert(tk.END, text), self.result_box.see(tk.END))) - except Exception: - # 在极少数情况下（例如 result_box 已销毁），捕获异常避免崩溃 - pass - - # ------------------------- - # 单端口测试 - # ------------------------- - def test_connect(self, ip: str, port: int, timeout: float = 1.0) -> bool: - """ - 立即测试单个 ip:port 是否可以 TCP 连接。 - - 返回 True（可连通）或 False（不可连通） - - 同时把结果写入 result_box（通过主线程调度） - """ - addr = (ip, int(port)) - status = False - try: - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.settimeout(timeout) - sock.connect(addr) - status = True - sock.close() - except Exception: - status = False - - # 输出结果（在 GUI 中显示） - self._append_text(f"{ip}:{port} {'✅ 开放' if status else '❌ 关闭/不可达'}\n") - return status - - # ------------------------- - # 并发单端口任务（内部使用） - # ------------------------- - def _scan_single_port(self, ip: str, port: int, timeout: float) -> str: - """ - 线程池中运行的单个端口检测任务，返回一行结果字符串。 - 任务必须尽量短小（快速返回），并在开始前检查 stop_flag。 - """ - if self._stop_flag: - return "" # 为空表示不输出 + def _scan(self, hosts: list[str], ports: list[int], options: ScanOptions, title: str) -> None: + stats = ScanStats(total=len(hosts) * len(ports)) + resolved_hosts = [] try: - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - sock.settimeout(timeout) - sock.connect((ip, port)) - sock.close() - result = f"{ip}:{port} ✅ 开放\n" - # 记录到本地开放端口列表（线程安全地追加） - self._open_ports.append(port) - except Exception: - result = f"{ip}:{port} ❌ 关闭/不可达\n" + for host in hosts: + if self.stop_event.is_set(): + break + info = resolve_host(host) + if info["ok"]: + resolved_hosts.append(info) + suffix = "" if info["resolved_ip"] == host else f" -> {info['resolved_ip']}" + self.output(f"解析: {host}{suffix} ({info['ip_version']})\n", "muted") + else: + self._record_host_error(host, info["error"], len(ports), stats) - # 更新进度计数（最好在主线程更新显示，这里在子线程更新计数） - self._done += 1 - return result - - # ------------------------- - # 并发扫描（范围或列表） - # ------------------------- - def start_range_scan(self, ip: str, start_port: int, end_port: int, timeout: float = 0.8, max_workers: int = 200): - """ - 并发扫描端口范围 [start_port, end_port]（含两端）。 - 结果实时写入 result_box。单次扫描在后台线程中运行（不会阻塞主线程）。 - """ - # 参数校验（基本） - try: - start_port = int(start_port); end_port = int(end_port) - except Exception: - messagebox.showwarning("输入错误", "起始端口和结束端口必须为整数") - return - if start_port < 1 or end_port > 65535 or start_port > end_port: - messagebox.showwarning("输入错误", "端口范围不合法（1-65535 且 起始<=结束）") - return - - ports = list(range(start_port, end_port + 1)) - self.start_list_scan(ip, ports, timeout=timeout, max_workers=max_workers) - - def start_list_scan(self, ip: str, ports: Iterable[int], timeout: float = 0.8, max_workers: int = 200): - """ - 并发扫描指定的端口列表。 - - ip: 目标 IP（字符串） - - ports: 可迭代的端口集合（如 list、range 等） - - timeout: 单端口连接超时（秒） - - max_workers: 最大并发数（线程池大小） - """ - # 防止重复启动 - if self._scan_thread and self._scan_thread.is_alive(): - messagebox.showinfo("提示", "已有扫描任务在运行，请先停止后再启动新的扫描。") - return - - # 将 ports 转为列表并进行基本校验 - try: - ports_list = [int(p) for p in ports] - except Exception: - messagebox.showwarning("输入错误", "端口列表包含非法值") - return - if not ports_list: - messagebox.showwarning("输入错误", "端口列表为空") - return - for p in ports_list: - if p < 1 or p > 65535: - messagebox.showwarning("输入错误", f"端口 {p} 不在合法范围 1-65535") - return - - # 重置状态 - self._stop_flag = False - self._open_ports = [] - self._total = len(ports_list) - self._done = 0 - - # 清空 result_box 并输出起始信息（主线程调度） - self._append_text(f"开始并发端口扫描：目标 {ip}，共 {self._total} 个端口\n") - - # 后台线程用于管理线程池与结果收集，确保 GUI 不阻塞 - def manager(): - # 根据任务数自适应限制并发数 - actual_workers = max(1, min(max_workers, self._total)) - self._executor = concurrent.futures.ThreadPoolExecutor(max_workers=actual_workers) - - # 提交任务 - futures = {self._executor.submit(self._scan_single_port, ip, port, timeout): port for port in ports_list} - - try: - # as_completed 会在每个 future 完成时迭代返回 - for fut in concurrent.futures.as_completed(futures): - if self._stop_flag: - # 如果外部发出停止信号，尽量取消未开始的 future（cancel 返回 True 表示取消成功） - # 线程池中的任务可能已在运行, cancel 只能取消未开始的任务 - break + workers = min(options.workers, max(1, stats.total)) + pair_iter = iter((host_info, port) for host_info in resolved_hosts for port in ports) + futures = {} + exhausted = False + self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=workers) + while not self.stop_event.is_set() and (futures or not exhausted): + while not self.stop_event.is_set() and not exhausted and len(futures) < workers * 2: try: - line = fut.result() - except Exception as e: - line = f"{ip}:{futures.get(fut)} 错误: {e}\n" + host_info, port = next(pair_iter) + except StopIteration: + exhausted = True + break + futures[self.executor.submit(self._scan_port, host_info, port, options)] = (host_info, port) - if line: - # 把结果写回 GUI（通过 _append_text 安全调度） - self._append_text(line) + if not futures: + break - # 可选：显示进度（例如：done/total） - self._append_text(f"进度: {self._done}/{self._total}\n") + done, _pending = concurrent.futures.wait( + futures, + timeout=0.15, + return_when=concurrent.futures.FIRST_COMPLETED, + ) + for future in done: + context = futures.pop(future, None) + try: + item = future.result() + except Exception as exc: + host_info, port = context or ({"host": "", "resolved_ip": "", "ip_version": ""}, 0) + item = PortResult( + host=host_info["host"], + resolved_ip=host_info["resolved_ip"], + ip_version=host_info["ip_version"], + port=port, + service=service_name(port), + status="error", + status_text=STATUS_TEXT["error"], + error=str(exc), + ) + self._record_result(item, stats, options) - # 如果 stop_flag 已设，尝试取消尚未开始的任务 - if self._stop_flag: - for f in futures: - f.cancel() + if self.stop_event.is_set(): + for future in futures: + future.cancel() + finally: + if self.executor: + self.executor.shutdown(wait=False, cancel_futures=True) + self.executor = None - finally: - # 关闭线程池 - if self._executor: - self._executor.shutdown(wait=False) - self._executor = None + state = "已停止" if self.stop_event.is_set() else "已完成" + self.status(stats.snapshot(state)) + self.last_results = sorted(self.last_results, key=result_sort_key) + self._write_summary(title, stats, state) + self.done() - # 最终输出总结信息（开放端口列表） - if not self._stop_flag: - self._append_text("\n端口扫描完成。\n") - else: - self._append_text("\n端口扫描已停止。\n") + def _record_host_error(self, host: str, error: str, port_count: int, stats: ScanStats) -> None: + item = PortResult( + host=host, + resolved_ip="", + ip_version="", + port=0, + service="", + status="dns_error", + status_text=STATUS_TEXT["dns_error"], + error=error, + ) + stats.record("dns_error", port_count) + self.last_results.append(item.to_dict()) + self.result(item.to_dict()) + self.output(f"{host} 解析失败: {error}\n", "warning") + self.status(stats.snapshot("运行中")) - if self._open_ports: - self._append_text(f"开放端口: {sorted(self._open_ports)}\n") - else: - self._append_text("未发现开放端口。\n") + def _record_result(self, item: PortResult, stats: ScanStats, options: ScanOptions) -> None: + stats.record(item.status) + row = item.to_dict() + self.last_results.append(row) + self.result(row) - # 启动后台管理线程 - self._scan_thread = threading.Thread(target=manager, daemon=True) - self._scan_thread.start() + should_print = options.show_closed or item.status == "open" or item.status in {"dns_error", "error"} + if should_print: + tag = "success" if item.status == "open" else "warning" if item.status != "closed" else None + latency = f"{item.latency_ms:.0f} ms" if item.latency_ms else "-" + detail = item.banner or item.error + detail = f" {detail}" if detail else "" + self.output( + f"[{stats.scanned}/{stats.total}] {item.host}:{item.port:<5} " + f"{item.service:<10} {item.status_text:<6} {latency}{detail}\n", + tag, + ) - # ------------------------- - # 停止扫描 - # ------------------------- - def stop_scan(self): - """ - 请求停止正在进行的扫描：设置停止标志并尝试关闭线程池。 - - 已提交并正在运行的 socket.connect 调用无法被立即中断（但后续任务会被取消） - - stop_scan 尽快返回，实际终止需要等待正在运行的任务完成或超时 - """ - if not (self._scan_thread and self._scan_thread.is_alive()): - messagebox.showinfo("提示", "当前没有正在运行的扫描任务。") - return + self.status(stats.snapshot("运行中")) - self._append_text("\n正在停止扫描，请稍候...\n") - self._stop_flag = True + def _scan_port(self, host_info: dict, port: int, options: ScanOptions) -> PortResult: + if self.stop_event.is_set(): + return PortResult( + host=host_info["host"], + resolved_ip=host_info["resolved_ip"], + ip_version=host_info["ip_version"], + port=port, + service=service_name(port), + status="cancelled", + status_text=STATUS_TEXT["cancelled"], + ) - # 尝试立即关闭线程池（不等待正在运行任务） - if self._executor: - try: - self._executor.shutdown(wait=False) - except Exception: - pass + started = time.perf_counter() + timeout = options.timeout_ms / 1000 + try: + with socket.create_connection((host_info["resolved_ip"], port), timeout=timeout) as sock: + elapsed = (time.perf_counter() - started) * 1000 + banner = self._read_banner(sock, host_info["host"], port, timeout) if options.banner_probe else "" + return PortResult( + host=host_info["host"], + resolved_ip=host_info["resolved_ip"], + ip_version=host_info["ip_version"], + port=port, + service=service_name(port), + status="open", + status_text=STATUS_TEXT["open"], + latency_ms=elapsed, + banner=banner, + ) + except socket.timeout: + return self._closed_result(host_info, port, "timeout", "连接超时") + except OSError as exc: + status = classify_os_error(exc) + return self._closed_result(host_info, port, status, clean_error(exc)) + + def _closed_result(self, host_info: dict, port: int, status: str, error: str) -> PortResult: + return PortResult( + host=host_info["host"], + resolved_ip=host_info["resolved_ip"], + ip_version=host_info["ip_version"], + port=port, + service=service_name(port), + status=status, + status_text=STATUS_TEXT.get(status, STATUS_TEXT["error"]), + error=error, + ) + + def _read_banner(self, sock: socket.socket, host: str, port: int, timeout: float) -> str: + try: + sock.settimeout(min(max(timeout, 0.25), 1.0)) + if port in HTTP_BANNER_PORTS: + request = f"HEAD / HTTP/1.0\r\nHost: {host}\r\nConnection: close\r\n\r\n" + sock.sendall(request.encode("ascii", errors="ignore")) + data = sock.recv(256) + except Exception: + return "" + text = data.decode("utf-8", errors="replace") + text = re.sub(r"\s+", " ", text.replace("\x00", " ")).strip() + return text[:160] + + def stop_scan(self) -> None: + if not self.is_scanning(): + raise RuntimeError("当前没有正在运行的端口扫描") + self.stop_event.set() + if self.executor: + self.executor.shutdown(wait=False, cancel_futures=True) + self.output("\n正在停止端口扫描...\n", "warning") + + def export_results(self, path: str) -> None: + if not self.last_results: + raise RuntimeError("还没有可导出的端口扫描结果") + fields = [ + "host", + "resolved_ip", + "ip_version", + "port", + "service", + "status", + "status_text", + "latency_ms", + "banner", + "error", + "checked_at", + ] + with open(path, "w", newline="", encoding="utf-8-sig") as file: + writer = csv.DictWriter(file, fieldnames=fields) + writer.writeheader() + for row in self.last_results: + writer.writerow({field_name: row.get(field_name, "") for field_name in fields}) + + def open_ports_summary(self) -> str: + open_items = [item for item in self.last_results if item.get("status") == "open"] + if not open_items: + return "" + grouped: dict[str, list[str]] = {} + for item in sorted(open_items, key=result_sort_key): + grouped.setdefault(item["host"], []).append(str(item["port"])) + return "\n".join(f"{host}: {', '.join(ports)}" for host, ports in grouped.items()) + + def describe_options(self, options: ScanOptions) -> str: + closed = "显示" if options.show_closed else "只显示开放端口" + banner = "开启" if options.banner_probe else "关闭" + return f"超时: {options.timeout_ms}ms 并发: {options.workers} 输出: {closed} Banner 探测: {banner}\n\n" + + def normalize_options(self, options: Optional[dict]) -> ScanOptions: + options = options or {} + return ScanOptions( + timeout_ms=clamp_int(options.get("timeout_ms", 800), 100, 60000, "超时"), + workers=clamp_int(options.get("workers", 128), 1, 512, "并发数"), + show_closed=bool(options.get("show_closed", False)), + banner_probe=bool(options.get("banner_probe", False)), + ) + + def _write_summary(self, title: str, stats: ScanStats, state: str) -> None: + snapshot = stats.snapshot(state) + self.output(f"\n==== {title}统计 ====\n", "muted") + self.output( + f"状态: {state} 已扫: {snapshot['scanned']}/{snapshot['total']} " + f"开放: {snapshot['open']} 关闭: {snapshot['closed']} 超时: {snapshot['timeout']} " + f"不可达: {snapshot['unreachable']} 错误: {snapshot['error'] + snapshot['dns_error']} " + f"耗时: {snapshot['elapsed']:.1f}s\n", + "success" if snapshot["open"] else "warning", + ) + summary = self.open_ports_summary() + if summary: + self.output("开放端口汇总:\n" + summary + "\n", "success") - # ------------------------- - # 可选：返回当前扫描状态（供外部查询） - # ------------------------- def is_scanning(self) -> bool: - return bool(self._scan_thread and self._scan_thread.is_alive()) + return bool(self.scan_thread and self.scan_thread.is_alive()) + + +def normalize_ports(ports: Iterable[int]) -> list[int]: + unique = sorted({validate_port(port) for port in ports}) + if not unique: + raise ValueError("请输入至少一个端口") + return unique + + +def parse_scan_hosts(text: str) -> list[str]: + raw = text.strip() + if not raw: + raise ValueError("请输入扫描目标") + + targets = [] + for part in re.split(r"[,，;\s]+", raw): + item = part.strip() + if not item: + continue + if "/" in item: + network = ipaddress.ip_network(item, strict=False) + targets.extend(str(ip) for ip in network.hosts()) + elif re.match(r"^\d{1,3}(?:\.\d{1,3}){3}-\d{1,3}$", item): + prefix, tail = item.rsplit(".", 1) + start_text, end_text = tail.split("-", 1) + start = int(start_text) + end = int(end_text) + if start > end: + raise ValueError("IP 范围起始值不能大于结束值") + targets.extend(str(ipaddress.ip_address(f"{prefix}.{value}")) for value in range(start, end + 1)) + elif re.match(r"^\d{1,3}(?:\.\d{1,3}){3}-\d{1,3}(?:\.\d{1,3}){3}$", item): + start_text, end_text = item.split("-", 1) + start_ip = ipaddress.ip_address(start_text) + end_ip = ipaddress.ip_address(end_text) + if start_ip.version != end_ip.version: + raise ValueError("IP 范围两端必须是同一 IP 版本") + if int(start_ip) > int(end_ip): + raise ValueError("IP 范围起始值不能大于结束值") + targets.extend(str(ipaddress.ip_address(value)) for value in range(int(start_ip), int(end_ip) + 1)) + else: + targets.append(validate_host(item)) + + unique = list(dict.fromkeys(targets)) + if not unique: + raise ValueError("没有解析到有效目标") + return unique + + +def resolve_host(host: str) -> dict: + try: + parsed = ipaddress.ip_address(host) + return {"ok": True, "host": host, "resolved_ip": str(parsed), "ip_version": f"IPv{parsed.version}", "error": ""} + except ValueError: + pass + + try: + infos = socket.getaddrinfo(host, None, type=socket.SOCK_STREAM) + except socket.gaierror as exc: + return {"ok": False, "host": host, "resolved_ip": "", "ip_version": "", "error": clean_error(exc)} + + addresses = [] + for family, _socktype, _proto, _canonname, sockaddr in infos: + if family not in (socket.AF_INET, socket.AF_INET6): + continue + ip = sockaddr[0] + if ip not in addresses: + addresses.append(ip) + + if not addresses: + return {"ok": False, "host": host, "resolved_ip": "", "ip_version": "", "error": "没有可用的 TCP 地址"} + + selected = addresses[0] + version = "IPv6" if ":" in selected else "IPv4" + return {"ok": True, "host": host, "resolved_ip": selected, "ip_version": version, "error": ""} + + +def service_name(port: int) -> str: + if port <= 0: + return "" + if port in COMMON_SERVICES: + return COMMON_SERVICES[port] + try: + return socket.getservbyport(port, "tcp").upper() + except OSError: + return "" + + +def classify_os_error(exc: OSError) -> str: + code = getattr(exc, "winerror", None) or getattr(exc, "errno", None) + text = str(exc).lower() + if code in {10061, 111, 61} or "refused" in text or "拒绝" in text: + return "closed" + if code in {10051, 10064, 10065, 101, 113} or "unreachable" in text or "不可达" in text: + return "unreachable" + if "timed out" in text or "超时" in text: + return "timeout" + return "error" + + +def clean_error(exc: BaseException) -> str: + if isinstance(exc, OSError): + return exc.strerror or str(exc) + return str(exc) + + +def result_sort_key(item: dict) -> tuple: + host_key = item.get("resolved_ip") or item.get("host") or "" + try: + host_key = f"{int(ipaddress.ip_address(host_key)):039d}" + except ValueError: + pass + return (host_key, int(item.get("port") or 0)) + + +def clamp_int(value, min_value: int, max_value: int, label: str) -> int: + try: + number = int(str(value).strip()) + except ValueError as exc: + raise ValueError(f"{label}必须是整数") from exc + if number < min_value or number > max_value: + raise ValueError(f"{label}必须在 {min_value}-{max_value} 之间") + return number diff --git a/core/Function/tracert_fun.py b/core/Function/tracert_fun.py index 43b6ea9..d9ff0b9 100644 --- a/core/Function/tracert_fun.py +++ b/core/Function/tracert_fun.py @@ -1,84 +1,69 @@ -import subprocess import threading -import tkinter as tk -from tkinter import scrolledtext, messagebox +from typing import Callable, Optional + +from core.Function.common import popen_hidden, validate_host + + +OutputCallback = Callable[[str, Optional[str]], None] +DoneCallback = Callable[[], None] class TracertFun: - def __init__(self, result_box: scrolledtext.ScrolledText): - self.result_box = result_box + def __init__(self, output: OutputCallback, done: Optional[DoneCallback] = None): + self.output = output + self.done = done or (lambda: None) self.process = None - self.stop_flag = False + self.stop_event = threading.Event() + self.worker = None - def _append_text(self, text: str): - """线程安全地输出到文本框""" - self.result_box.after(0, lambda: ( - self.result_box.insert(tk.END, text), - self.result_box.see(tk.END) - )) + def start_tracert(self, target: str, max_hops: int = 20, timeout_ms: int = 800) -> None: + target = validate_host(target) + max_hops = max(1, min(int(max_hops), 64)) + timeout_ms = max(100, min(int(timeout_ms), 10000)) - def start_tracert(self, target: str): - """开始追踪""" - if self.process: - messagebox.showwarning("警告", "⚠️ 正在运行，请先停止再启动。") - return + if self.is_running(): + raise RuntimeError("路由追踪正在运行，请先停止当前任务") - if not target.strip(): - messagebox.showwarning("提示", "请输入目标地址！") - return + self.stop_event.clear() + command = ["tracert", "-d", "-w", str(timeout_ms), "-h", str(max_hops), target] + self.output(f"开始路由追踪: {target}，最大 {max_hops} 跳，超时 {timeout_ms}ms\n\n", "muted") + self.worker = threading.Thread(target=self._run, args=(command,), daemon=True) + self.worker.start() - cmd = f'tracert -d -w 500 -h 20 {target}' - self.stop_flag = False - - self._append_text(f"\n=== 开始追踪 {target} ===\n\n") - - thread = threading.Thread(target=self._run_tracert, args=(cmd,)) - thread.daemon = True - thread.start() - - def _run_tracert(self, cmd: str): - """执行 tracert 命令""" + def _run(self, command) -> None: try: - self.process = subprocess.Popen( - cmd, - shell=True, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - text=True, - creationflags=subprocess.CREATE_NO_WINDOW # 隐藏控制台窗口 - ) - + self.process = popen_hidden(command) + if not self.process.stdout: + return for line in self.process.stdout: - if self.stop_flag: + if self.stop_event.is_set(): break - self._append_text(line) - - except Exception as e: - self._append_text(f"\n❌ 错误: {e}\n") - + self.output(line, None) + except Exception as exc: + self.output(f"\n路由追踪失败: {exc}\n", "error") finally: - # 安全关闭进程 if self.process: try: self.process.terminate() except Exception: pass self.process = None - - if self.stop_flag: - self._append_text("\n=== 已停止追踪 ===\n") + if self.stop_event.is_set(): + self.output("\n路由追踪已停止\n", "warning") else: - self._append_text("\n--- 追踪结束 ---\n") + self.output("\n路由追踪完成\n", "success") + self.done() - def stop_tracert(self): - """停止追踪""" + def stop_tracert(self) -> None: + if not self.is_running(): + raise RuntimeError("当前没有正在运行的路由追踪") + self.stop_event.set() if self.process: - self.stop_flag = True try: self.process.terminate() except Exception: pass - self.process = None - self._append_text("\n=== 已手动停止追踪 ===\n") - else: - messagebox.showinfo("提示", "当前没有正在运行的追踪任务。") + self.output("\n正在停止路由追踪...\n", "warning") + + def is_running(self) -> bool: + return bool(self.worker and self.worker.is_alive()) diff --git a/core/logger_config.py b/core/logger_config.py index 5282afa..fbd1144 100644 --- a/core/logger_config.py +++ b/core/logger_config.py @@ -2,25 +2,25 @@ import logging import os import sys -def get_base_dir(): - """获取程序真实所在目录，兼容开发与打包""" - if getattr(sys, 'frozen', False): # 打包后的 exe - return os.path.dirname(sys.executable) - else: # 普通 Python 运行 - return os.path.dirname(os.path.abspath(sys.argv[0])) -def setup_logger(): - base_dir = get_base_dir() - log_file = os.path.join(base_dir, 'app.log') # 直接放在 main 同级目录 +def get_base_dir() -> str: + if getattr(sys, "frozen", False): + return os.path.dirname(sys.executable) + return os.path.dirname(os.path.abspath(sys.argv[0])) + + +def setup_logger() -> logging.Logger: + log_file = os.path.join(get_base_dir(), "app.log") logging.basicConfig( level=logging.INFO, - format='%(asctime)s [%(levelname)s] %(message)s', + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", handlers=[ - logging.FileHandler(log_file, encoding='utf-8'), - logging.StreamHandler() - ] + logging.FileHandler(log_file, encoding="utf-8"), + logging.StreamHandler(), + ], ) - logging.info("日志系统初始化完成，日志文件路径：%s", log_file) - return logging.getLogger(__name__) + logger = logging.getLogger(__name__) + logger.info("日志系统初始化完成: %s", log_file) + return logger diff --git a/core/ui/basic_ui.py b/core/ui/basic_ui.py index a1e3797..ab126b6 100644 --- a/core/ui/basic_ui.py +++ b/core/ui/basic_ui.py @@ -1,112 +1,71 @@ -import os import tkinter as tk -from tkinter import ttk, filedialog, messagebox +from tkinter import ttk -import logging -logger = logging.getLogger(__name__) class BasicUI: + """旧版页面的兼容辅助类。 + + 新界面主要使用 core.ui.components；保留这个类是为了兼容后续可能仍引用 + add_input/add_combobox/add_button 的小工具页。 + """ def add_input( - self, parent, label_text, - row, col=0, inivar="", - label_width=8, entry_width=20, colspan=1, sticky='w' - ): - """ - 添加一个带文本标签和输入框的组合控件，并返回 (StringVar, Frame) 以便后续控制。 - - parent: 父容器 - - label: 标签文字 - - row, col: 放置在父容器的 grid 行列 - - inivar: 初始值 - - entry_width: 输入框宽度 - - colspan: 该组控件在父容器上跨越的列数 - - sticky: 对齐方式（默认左对齐） - """ + self, + parent, + label_text, + row, + col=0, + inivar="", + label_width=8, + entry_width=20, + colspan=1, + sticky="w", + ): group_frame = ttk.Frame(parent) group_frame.grid(row=row, column=col, columnspan=colspan, sticky=sticky, padx=5, pady=3) - # 标签 - label = ttk.Label(group_frame, text=label_text + "：", width=label_width, anchor='w').grid(row=0, column=0, sticky='w', padx=(0, 5)) + label = ttk.Label(group_frame, text=f"{label_text}:", width=label_width, anchor="w") + label.grid(row=0, column=0, sticky="w", padx=(0, 5)) - # 输入框 var = tk.StringVar(value=inivar) entry = ttk.Entry(group_frame, textvariable=var, width=entry_width) - entry.grid(row=0, column=1, sticky='w') + entry.grid(row=0, column=1, sticky="w") - return { - "frame": group_frame, - "label": label, - "entry": entry, - "var": var - } + return {"frame": group_frame, "label": label, "entry": entry, "var": var} def add_combobox( - self, parent, label_text, row, col=0, - listbox=[], inivar=0, width=17, colspan=1, - sticky='w', label_width=8, state="readonly" + self, + parent, + label_text, + row, + col=0, + listbox=None, + inivar=0, + width=17, + colspan=1, + sticky="w", + label_width=8, + state="readonly", ): - """ - 创建一组 [标签 + 下拉框] 控件。 - 返回 dict，方便外部单独或统一控制。 - - - values: 下拉选项列表 - - default: 初始值（可选） - - state: "readonly" 表示只能从列表选，"normal" 可手动输入 - """ + values = listbox or [] frame = ttk.Frame(parent) frame.grid(row=row, column=col, columnspan=colspan, sticky=sticky, padx=5, pady=3) - # 标签 - label = ttk.Label( - frame, - text=label_text + "：", - width=label_width, - anchor='w' - ) - label.grid(row=0, column=0, sticky='w', padx=(0, 5)) + label = ttk.Label(frame, text=f"{label_text}:", width=label_width, anchor="w") + label.grid(row=0, column=0, sticky="w", padx=(0, 5)) - # 变量 + 下拉框 var = tk.StringVar() - combobox = ttk.Combobox( - frame, - textvariable=var, - values=listbox, - width=width, - state=state - ) - combobox.grid(row=0, column=1, sticky='w') - if inivar>=0: - var.set(listbox[inivar]) + combobox = ttk.Combobox(frame, textvariable=var, values=values, width=width, state=state) + combobox.grid(row=0, column=1, sticky="w") + if values and inivar >= 0: + var.set(values[inivar]) - return { - "frame": frame, - "label": label, - "combobox": combobox, - "var": var - } + return {"frame": frame, "label": label, "combobox": combobox, "var": var} - def add_button( - self, parent, button_text, - row, col=0, command="", - width=5, colspan=1, sticky='w' - ): - """ - 添加一个按钮，并返回 (StringVar, Frame) 以便后续控制。 - - parent: 父容器 - - button_text: 按钮文本 - - row, col: 放置在父容器的 grid 行列 - - command: 调用的函数 - - width: 按钮宽度 - - colspan: 该组控件在父容器上跨越的列数 - - sticky: 对齐方式（默认左对齐） - """ - group_frame = ttk.Frame(parent) - group_frame.grid(row=row, column=col, columnspan=colspan, sticky=sticky, padx=5, pady=3) + def add_button(self, parent, button_text, row, col=0, command=None, width=5, colspan=1, sticky="w"): + group_frame = ttk.Frame(parent) + group_frame.grid(row=row, column=col, columnspan=colspan, sticky=sticky, padx=5, pady=3) - btn = ttk.Button(group_frame, text=button_text, command=command, width=width) - btn.grid(row=0, column=0, sticky='w') - return { - "frame": group_frame, - "btn": btn, - } - \ No newline at end of file + btn = ttk.Button(group_frame, text=button_text, command=command, width=width) + btn.grid(row=0, column=0, sticky="w") + return {"frame": group_frame, "btn": btn} diff --git a/core/ui/components.py b/core/ui/components.py new file mode 100644 index 0000000..e5d5afd --- /dev/null +++ b/core/ui/components.py @@ -0,0 +1,144 @@ +import tkinter as tk +from tkinter import scrolledtext, ttk +from typing import Callable, Optional + +from core.ui.theme import COLORS, FONT_MONO, FONT_SECTION, FONT_SMALL + + +class Console: + def __init__(self, parent, height: int = 16): + self.widget = scrolledtext.ScrolledText( + parent, + height=height, + wrap="word", + bg=COLORS["console_bg"], + fg=COLORS["console_fg"], + insertbackground=COLORS["console_fg"], + selectbackground="#2b4a6f", + relief="flat", + borderwidth=0, + font=FONT_MONO, + padx=14, + pady=12, + ) + self.widget.tag_config("muted", foreground=COLORS["console_muted"]) + self.widget.tag_config("success", foreground="#86efac") + self.widget.tag_config("warning", foreground="#fde68a") + self.widget.tag_config("error", foreground="#fca5a5") + + def grid(self, **kwargs): + self.widget.grid(**kwargs) + + def pack(self, **kwargs): + self.widget.pack(**kwargs) + + def clear(self) -> None: + self.widget.delete("1.0", tk.END) + + def write(self, text: str, tag: Optional[str] = None) -> None: + def append(): + self.widget.insert(tk.END, text, tag) + self.widget.see(tk.END) + + self.widget.after(0, append) + + +class Page(ttk.Frame): + def __init__(self, parent, title: str, subtitle: str): + super().__init__(parent, style="Panel.TFrame") + self.columnconfigure(0, weight=1) + self.rowconfigure(1, weight=1) + + header = ttk.Frame(self, style="Panel.TFrame") + header.grid(row=0, column=0, sticky="ew", padx=28, pady=(24, 8)) + header.columnconfigure(0, weight=1) + + ttk.Label(header, text=title, style="Title.TLabel").grid(row=0, column=0, sticky="w") + ttk.Label(header, text=subtitle, style="Muted.TLabel").grid(row=1, column=0, sticky="w", pady=(4, 0)) + + body_shell = ttk.Frame(self, style="Panel.TFrame") + body_shell.grid(row=1, column=0, sticky="nsew", padx=28, pady=(4, 20)) + body_shell.columnconfigure(0, weight=1) + body_shell.rowconfigure(0, weight=1) + + self._canvas = tk.Canvas(body_shell, bg=COLORS["panel"], highlightthickness=0, bd=0) + self._canvas.grid(row=0, column=0, sticky="nsew") + + scrollbar = ttk.Scrollbar(body_shell, orient="vertical", command=self._canvas.yview) + scrollbar.grid(row=0, column=1, sticky="ns", padx=(8, 0)) + self._canvas.configure(yscrollcommand=scrollbar.set) + + self.body = ttk.Frame(self._canvas, style="Panel.TFrame") + self.body.columnconfigure(0, weight=1) + self._body_window = self._canvas.create_window((0, 0), window=self.body, anchor="nw") + + self.body.bind("", self._update_scroll_region) + self._canvas.bind("", self._resize_body) + self._bind_mousewheel(self._canvas) + + def section(self, title: str, row: int, columns: int = 4): + frame = tk.Frame(self.body, bg=COLORS["panel"], highlightbackground=COLORS["border"], highlightthickness=1) + frame.grid(row=row, column=0, sticky="ew", pady=(0, 14)) + for col in range(columns): + frame.columnconfigure(col, weight=1) + + ttk.Label(frame, text=title, style="Section.TLabel").grid( + row=0, column=0, columnspan=columns, sticky="w", padx=18, pady=(14, 8) + ) + return frame + + def _update_scroll_region(self, _event=None) -> None: + self._canvas.configure(scrollregion=self._canvas.bbox("all")) + + def _resize_body(self, event) -> None: + self._canvas.itemconfigure(self._body_window, width=event.width) + + def _bind_mousewheel(self, widget) -> None: + widget.bind("", lambda _event: widget.bind_all("", self._on_mousewheel)) + widget.bind("", lambda _event: widget.unbind_all("")) + + def _on_mousewheel(self, event) -> None: + if not self.winfo_ismapped(): + return + self._canvas.yview_scroll(int(-1 * (event.delta / 120)), "units") + + +def field(parent, label: str, row: int, column: int, value: str = "", width: int = 24, colspan: int = 1): + frame = ttk.Frame(parent, style="Panel.TFrame") + frame.grid(row=row, column=column, columnspan=colspan, sticky="ew", padx=18, pady=(4, 14)) + frame.columnconfigure(0, weight=1) + + ttk.Label(frame, text=label, style="Muted.TLabel").grid(row=0, column=0, sticky="w", pady=(0, 5)) + var = tk.StringVar(value=value) + entry = ttk.Entry(frame, textvariable=var, width=width) + entry.grid(row=1, column=0, sticky="ew") + return {"frame": frame, "var": var, "entry": entry} + + +def combo(parent, label: str, row: int, column: int, values=None, value: str = "", width: int = 24, colspan: int = 1): + values = values or [] + frame = ttk.Frame(parent, style="Panel.TFrame") + frame.grid(row=row, column=column, columnspan=colspan, sticky="ew", padx=18, pady=(4, 14)) + frame.columnconfigure(0, weight=1) + + ttk.Label(frame, text=label, style="Muted.TLabel").grid(row=0, column=0, sticky="w", pady=(0, 5)) + var = tk.StringVar(value=value) + control = ttk.Combobox(frame, textvariable=var, values=values, width=width, state="readonly") + control.grid(row=1, column=0, sticky="ew") + return {"frame": frame, "var": var, "combobox": control} + + +def action_bar(parent, row: int, columnspan: int = 4): + frame = ttk.Frame(parent, style="Panel.TFrame") + frame.grid(row=row, column=0, columnspan=columnspan, sticky="ew", padx=18, pady=(0, 16)) + return frame + + +def button(parent, text: str, command: Callable, style: str = "Secondary.TButton"): + btn = ttk.Button(parent, text=text, command=command, style=style) + btn.pack(side="left", padx=(0, 10)) + return btn + + +def set_entry_state(item, enabled: bool) -> None: + item["entry"].configure(state="normal" if enabled else "disabled") diff --git a/core/ui/tab_network.py b/core/ui/tab_network.py index 4676884..3b991f8 100644 --- a/core/ui/tab_network.py +++ b/core/ui/tab_network.py @@ -1,159 +1,249 @@ -import tkinter as tk -from tkinter import ttk, messagebox, scrolledtext import threading -import re +from tkinter import messagebox -import logging -from core.ui.basic_ui import BasicUI -from core.Function.network_fun import NetworkManager +from core.Function.network_fun import NetworkManager +from core.ui.components import Console, Page, action_bar, button, combo, field, set_entry_state -logger = logging.getLogger(__name__) - -class NetworkTab(ttk.Frame, BasicUI): - """网卡配置 Tab（兼容最新 NetworkManager）""" +class NetworkTab(Page): def __init__(self, parent): - super().__init__(parent) - - self.networkname_list = [] - self.networkconfig = [] - self.build_ui() - self.netmgr = NetworkManager(self.result_box) + super().__init__(parent, "网卡配置", "查看本机网卡信息，切换 DHCP，或写入静态 IPv4 / DNS 配置。") + self.body.rowconfigure(4, weight=1) + self.adapters = [] + self.profiles = {} - def build_ui(self): - self.create_iface_section() - self.create_config_section() - self.create_action_section() - self.create_output_section() + selector = self.section("网卡选择", 0, columns=4) + self.iface = combo(selector, "选择网卡", 1, 0, [], "", 44, colspan=2) + self.iface["combobox"].bind("<>", lambda _event: self.load_selected_adapter()) + self.status = field(selector, "连接状态", 1, 2, "", 14) + self.status["entry"].configure(state="disabled") + actions = action_bar(selector, 2, 4) + self.refresh_btn = button(actions, "刷新网卡", self.refresh_adapters, "Primary.TButton") - def refresh_allnetwork(self): - '''刷新网卡列表''' - self.networkname_list = [] - try: - self.networkconfigs = self.netmgr.get_network_info() - for config in self.networkconfigs: - self.networkname_list.append(config.get("name", "未知网卡")) - self.iface_cb['combobox']['values'] = self.networkname_list - if self.iface_cb['var'].get() == "" : - self.iface_cb['var'].set(self.networkname_list[0]) - self.result_box.insert(tk.END, f"获取网卡信息完成，共:{len(self.networkname_list)}个启用网卡\n") - self.refresh_network_callback() + identity = self.section("网卡信息", 1, columns=4) + self.description = field(identity, "设备描述", 1, 0, "", 46, colspan=2) + self.mac = field(identity, "MAC 地址", 1, 2, "", 24) + for item in (self.description, self.mac): + item["entry"].configure(state="disabled") - logger.info(f"获取网卡信息完成，共:{len(self.networkname_list)}个启用网卡") - logger.info(f"所有获取网卡信息：{self.networkconfigs}") - except Exception as e: - messagebox.showerror("错误", f"获取网卡信息失败: {e}") - logger.error(f"获取网卡信息失败: {e}") - self.networkconfigs = [] + config = self.section("IPv4 配置", 2, columns=4) + self.dhcp = combo(config, "地址模式", 1, 0, ["DHCP 自动获取", "静态手动配置"], "DHCP 自动获取", 18) + self.dhcp["combobox"].bind("<>", lambda _event: self.update_entry_state()) + self.ipv4 = field(config, "IPv4 地址", 1, 1, "", 18) + self.netmask = field(config, "子网掩码", 1, 2, "", 18) + self.gateway = field(config, "默认网关", 1, 3, "", 18) + self.dns1 = field(config, "首选 DNS", 2, 1, "", 18) + self.dns2 = field(config, "备用 DNS", 2, 2, "", 18) + config_actions = action_bar(config, 3, 4) + self.apply_btn = button(config_actions, "应用配置", self.apply_settings, "Primary.TButton") + self.reload_btn = button(config_actions, "重新读取", self.refresh_adapters, "Secondary.TButton") - # ---------------- UI 构建 ---------------- - def create_iface_section(self): - frame = ttk.LabelFrame(self, text="网卡选择", padding=8) - frame.pack(side='top', fill='x', padx=10, pady=6) + profiles = self.section("配置模板", 3, columns=4) + self.profile_name = field(profiles, "模板名称", 1, 0, "", 20) + self.profile_select = combo(profiles, "选择模板", 1, 1, [], "", 24) + profile_actions = action_bar(profiles, 2, 4) + self.save_profile_btn = button(profile_actions, "保存当前为模板", self.save_profile, "Primary.TButton") + self.apply_profile_btn = button(profile_actions, "套用模板到表单", self.apply_profile_to_form, "Secondary.TButton") + self.delete_profile_btn = button(profile_actions, "删除模板", self.delete_profile, "Danger.TButton") - self.iface_cb = self.add_combobox(frame, "选择网卡", row=0, col=0, listbox=[], width=42, inivar=-1) - self.iface_cb['combobox'].bind("<>", lambda e: self.refresh_network_callback()) - self.refresh_btn = self.add_button(frame, "刷新网卡列表", row=0, col=1, width=15, command=self.refresh_allnetwork) - self.description_entry = self.add_input(frame, "网卡名称", row=1, col=0, inivar="", entry_width=45) - self.description_entry['entry'].config(state='disabled') - self.mac_entry = self.add_input(frame, "MAC地址", row=1, col=1, inivar="", entry_width=35) - self.mac_entry['entry'].config(state='disabled') + output = self.section("输出控制台", 4, columns=1) + output.rowconfigure(1, weight=1) + output.columnconfigure(0, weight=1) + self.console = Console(output, height=12) + self.console.grid(row=1, column=0, sticky="nsew", padx=18, pady=(0, 18)) - def create_config_section(self): - frame = ttk.LabelFrame(self, text="IP 配置（编辑后点击应用）", padding=8) - frame.pack(side='top', fill='x', padx=10, pady=6) + self.netmgr = NetworkManager(self.write) + self.load_profiles() + self.after(250, self.refresh_adapters) - self.ip_entry = self.add_input(frame, "IPv4 地址", row=0, col=0, inivar="", entry_width=20) - self.mask_entry = self.add_input(frame, "子网掩码", row=0, col=1, inivar="", entry_width=20) - self.gw_entry = self.add_input(frame, "默认网关", row=0, col=2, inivar="", entry_width=20) - #self.dns1_entry = self.add_input(frame, "主 DNS", row=1, col=0, inivar="", entry_width=20) - #self.dns2_entry = self.add_input(frame, "备选 DNS", row=1, col=1, inivar="", entry_width=20) - self.dhcp_cb = self.add_combobox(frame, "自动获取", row=2, col=0, listbox=["是","否"], width=5, inivar=-1) - self.dhcp_cb['combobox'].bind("<>", lambda e: self.dhcp_entry_state()) + def write(self, text, tag=None): + self.console.write(text, tag) - def create_action_section(self): - frame = ttk.LabelFrame(self, text="修改操作", padding=8) - frame.pack(side='top', fill='x', padx=10, pady=6) + def refresh_adapters(self, clear=True): + self.refresh_btn.configure(state="disabled") + self.reload_btn.configure(state="disabled") + if clear: + self.console.clear() + self.write("正在读取本机网卡信息...\n", "muted") - self.apply_btn = self.add_button(frame, "应用修改", row=0, col=0, width=10, command=self.apply_btn_callback) - self.refresh_info_btn = self.add_button(frame, "刷新当前信息", row=0, col=2, width=15, command=self.refresh_allnetwork) + def worker(): + try: + adapters = self.netmgr.get_network_info() + self.after(0, lambda: self.on_adapters_loaded(adapters)) + except Exception as exc: + self.after(0, lambda: self.on_refresh_failed(exc)) - def create_output_section(self): - frame = ttk.LabelFrame(self, text="输出信息", padding=6) - frame.pack(side='top', fill='both', expand=True, padx=10, pady=6) + threading.Thread(target=worker, daemon=True).start() - self.result_box = scrolledtext.ScrolledText(frame, width=100, height=12) - self.result_box.pack(fill='both', expand=True, padx=4, pady=4) - - - # ---------------- 事件回调 ---------------- - def refresh_network_callback(self): - '''刷新当前网卡信息''' - # 刷新所有所有网卡信息 - #self.refresh_allnetwork() - # 获取选中的网卡信息 - iface_name = self.iface_cb['var'].get() - config = self.get_network_config(iface_name) - # 不可修改配置 - self.description_entry['var'].set(config.get("description", "")) - self.mac_entry['var'].set(config.get("mac", "")) - # 可修改配置 - self.ip_entry['var'].set(config.get("ipv4", "")) - self.mask_entry['var'].set(config.get("netmask", "")) - self.gw_entry['var'].set(config.get("gateway", "")) - #self.dns1_entry['var'].set(config.get("dns1", "")) - #self.dns2_entry['var'].set(config.get("dns2", "")) - if config.get("dhcp_enabled", "") == "是": - self.dhcp_cb['var'].set("是") - elif config.get("dhcp_enabled", "") == "否": - self.dhcp_cb['var'].set("否") + def on_adapters_loaded(self, adapters): + self.adapters = adapters + names = [adapter["name"] for adapter in adapters] + self.iface["combobox"]["values"] = names + if names: + current = self.iface["var"].get() + self.iface["var"].set(current if current in names else names[0]) + self.load_selected_adapter() + self.write(f"读取完成，共发现 {len(names)} 个网卡\n", "success") else: - self.dhcp_cb['var'].set("") - self.dhcp_entry_state() - self.get_network_settings() - self.output_network_settings() + self.write("未发现可用网卡\n", "warning") + self.refresh_btn.configure(state="normal") + self.reload_btn.configure(state="normal") - def get_network_config(self, iface_name): - '''根据网卡名称获取对应的配置信息''' - for config in self.networkconfigs: - if config.get("name") == iface_name: - return config + def on_refresh_failed(self, exc): + self.write(f"读取网卡失败: {exc}\n", "error") + self.refresh_btn.configure(state="normal") + self.reload_btn.configure(state="normal") + messagebox.showerror("读取网卡失败", str(exc)) + + def load_selected_adapter(self): + adapter = self.current_adapter() + if not adapter: + return + + self.status["entry"].configure(state="normal") + self.description["entry"].configure(state="normal") + self.mac["entry"].configure(state="normal") + + self.status["var"].set(adapter.get("status", "")) + self.description["var"].set(adapter.get("description", "")) + self.mac["var"].set(adapter.get("mac", "")) + self.ipv4["var"].set(adapter.get("ipv4", "")) + self.netmask["var"].set(adapter.get("netmask", "")) + self.gateway["var"].set(adapter.get("gateway", "")) + self.dns1["var"].set(adapter.get("dns1", "")) + self.dns2["var"].set(adapter.get("dns2", "")) + self.dhcp["var"].set("DHCP 自动获取" if adapter.get("dhcp_enabled") else "静态手动配置") + + self.status["entry"].configure(state="disabled") + self.description["entry"].configure(state="disabled") + self.mac["entry"].configure(state="disabled") + self.update_entry_state() + self.write_current_adapter(adapter) + + def current_adapter(self): + name = self.iface["var"].get() + for adapter in self.adapters: + if adapter.get("name") == name: + return adapter return None - def dhcp_entry_state(self): - '''根据 DHCP 选择框设置输入框状态''' - if self.dhcp_cb['var'].get() == "是": - # 禁用手动输入 - self.ip_entry['entry'].config(state='disabled') - self.mask_entry['entry'].config(state='disabled') - self.gw_entry['entry'].config(state='disabled') - else: - # 启用手动输入 - self.ip_entry['entry'].config(state='normal') - self.mask_entry['entry'].config(state='normal') - self.gw_entry['entry'].config(state='normal') + def update_entry_state(self): + static = self.dhcp["var"].get() == "静态手动配置" + for item in (self.ipv4, self.netmask, self.gateway, self.dns1, self.dns2): + set_entry_state(item, static) - def get_network_settings(self): - '''获取当前输入的网卡配置信息''' - self.networkconfig = { - "name": re.sub(r"^(.*?(适配器))\s*", "", self.iface_cb['var'].get()).strip(), - "ipv4": self.ip_entry['var'].get(), - "netmask": self.mask_entry['var'].get(), - "gateway": self.gw_entry['var'].get(), - #"dns1": self.dns1_entry['var'].get(), - #"dns2": self.dns2_entry['var'].get(), - "dhcp_enabled": self.dhcp_cb['var'].get() == "是" + def write_current_adapter(self, adapter): + self.write("\n当前网卡:\n", "muted") + rows = [ + ("名称", adapter.get("name", "")), + ("描述", adapter.get("description", "")), + ("状态", adapter.get("status", "")), + ("速率", adapter.get("link_speed", "")), + ("接口索引", adapter.get("interface_index", "")), + ("IPv4", adapter.get("ipv4", "")), + ("IPv6", ", ".join(adapter.get("ipv6", []))), + ("掩码", adapter.get("netmask", "")), + ("网关", adapter.get("gateway", "")), + ("DNS", ", ".join(value for value in (adapter.get("dns1", ""), adapter.get("dns2", "")) if value)), + ("DHCP", "是" if adapter.get("dhcp_enabled") else "否"), + ("DHCP 服务器", adapter.get("dhcp_server", "")), + ("租约获取", adapter.get("dhcp_lease_obtained", "")), + ("租约过期", adapter.get("dhcp_lease_expires", "")), + ] + for key, value in rows: + self.write(f" {key}: {value}\n", "muted") + + def collect_settings(self): + return { + "name": self.iface["var"].get(), + "dhcp_enabled": self.dhcp["var"].get() == "DHCP 自动获取", + "ipv4": self.ipv4["var"].get(), + "netmask": self.netmask["var"].get(), + "gateway": self.gateway["var"].get(), + "dns1": self.dns1["var"].get(), + "dns2": self.dns2["var"].get(), } - def output_network_settings(self): - '''输出当前网卡配置信息到日志框''' - self.result_box.insert(tk.END, f"当前网卡配置：\n") - for key, value in self.networkconfig.items(): - self.result_box.insert(tk.END, f" {key}: {value}\n") - self.result_box.see(tk.END) + def apply_settings(self): + settings = self.collect_settings() + if not settings["name"]: + messagebox.showwarning("无法应用配置", "请先选择网卡") + return - def apply_btn_callback(self): - '''应用当前网卡配置信息''' - self.get_network_settings() - self.netmgr.set_network_info(self.networkconfig) - #self.refresh_allnetwork() \ No newline at end of file + self.apply_btn.configure(state="disabled") + self.write("\n开始应用配置。修改网卡通常需要管理员权限。\n", "warning") + + def worker(): + try: + self.netmgr.set_network_info(settings) + self.after(0, self.on_apply_success) + except Exception as exc: + self.after(0, lambda: self.on_apply_failed(exc)) + + threading.Thread(target=worker, daemon=True).start() + + def on_apply_success(self): + self.apply_btn.configure(state="normal") + self.write("配置应用完成，正在刷新网卡信息...\n", "success") + self.refresh_adapters(clear=False) + + def on_apply_failed(self, exc): + self.apply_btn.configure(state="normal") + self.write(f"配置应用失败: {exc}\n", "error") + messagebox.showerror("配置应用失败", f"{exc}\n\n请确认程序已用管理员权限运行。") + + def load_profiles(self): + try: + self.profiles = self.netmgr.load_profiles() + except Exception as exc: + self.profiles = {} + self.write(f"读取模板失败: {exc}\n", "warning") + names = sorted(self.profiles.keys()) + self.profile_select["combobox"]["values"] = names + if names and not self.profile_select["var"].get(): + self.profile_select["var"].set(names[0]) + + def save_profile(self): + try: + name = self.profile_name["var"].get().strip() + if not name: + adapter = self.current_adapter() + name = adapter.get("name", "未命名模板") if adapter else "未命名模板" + self.netmgr.save_profile(name, self.collect_settings()) + self.profile_name["var"].set(name) + self.profile_select["var"].set(name) + self.load_profiles() + self.write(f"已保存配置模板: {name}\n", "success") + except Exception as exc: + messagebox.showwarning("保存模板失败", str(exc)) + + def apply_profile_to_form(self): + name = self.profile_select["var"].get() + profile = self.profiles.get(name) + if not profile: + messagebox.showinfo("提示", "请选择要套用的模板") + return + + self.dhcp["var"].set("DHCP 自动获取" if profile.get("dhcp_enabled") else "静态手动配置") + self.ipv4["var"].set(profile.get("ipv4", "")) + self.netmask["var"].set(profile.get("netmask", "")) + self.gateway["var"].set(profile.get("gateway", "")) + self.dns1["var"].set(profile.get("dns1", "")) + self.dns2["var"].set(profile.get("dns2", "")) + self.update_entry_state() + self.write(f"已套用模板到表单: {name}\n", "success") + + def delete_profile(self): + name = self.profile_select["var"].get() + if not name: + messagebox.showinfo("提示", "请选择要删除的模板") + return + if not messagebox.askyesno("确认删除模板", f"确定删除配置模板“{name}”吗？"): + return + try: + self.netmgr.delete_profile(name) + self.profile_select["var"].set("") + self.load_profiles() + self.write(f"已删除配置模板: {name}\n", "success") + except Exception as exc: + messagebox.showwarning("删除模板失败", str(exc)) diff --git a/core/ui/tab_ping.py b/core/ui/tab_ping.py index 7ce2e5d..10baf18 100644 --- a/core/ui/tab_ping.py +++ b/core/ui/tab_ping.py @@ -1,122 +1,234 @@ +import re +import threading import tkinter as tk -from tkinter import ttk, filedialog, messagebox, scrolledtext +from tkinter import filedialog, messagebox, ttk -from core.ui.basic_ui import BasicUI +from core.Function.network_fun import NetworkManager from core.Function.ping_fun import PingFun +from core.ui.components import Console, Page, action_bar, button, combo, field -import logging -logger = logging.getLogger(__name__) -class PingTab(ttk.Frame, BasicUI): +DEFAULT_SOURCE = "默认路由" +IP_PATTERN = re.compile(r"(? 255: - messagebox.showwarning("输入错误", "起始地址最小为1，结束地址最大为255！") - return - if start > end: - messagebox.showwarning("输入错误", "起始地址不能大于结束地址！") - return - if not net_prefix.endswith('.'): - messagebox.showwarning("输入错误", "网段必须以 '.' 结尾，例如：192.168.1.") - return - if local_ip == "": - logger.info(f"开始批量 Ping {net_prefix}{start} - {net_prefix}{end}") - else: - logger.info(f"开始由{local_ip} 批量 Ping {net_prefix}{start} - {net_prefix}{end}") - - self.ping_fun.start_batch_ping(net_prefix, start, end, local_ip=local_ip, callback=self.batchIP_ping_callback) - self.batchIP_startPing['btn'].config(state='disabled') + try: + self.ping_fun.export_batch_results(path) + self.write(f"已导出结果: {path}\n", "success") + except Exception as exc: + messagebox.showwarning("导出失败", str(exc)) - def btn_batchIP_stopPing(self): - net_prefix = self.entry_batchIP_B['var'].get() - start = self.entry_batchIP_B_begin['var'].get() - end = self.entry_batchIP_B_end['var'].get() - local_ip = self.entry_batchIP_A['var'].get() - if local_ip == "": - logger.info(f"停止批量 Ping {net_prefix}{start} - {net_prefix}{end}") - else: - logger.info(f"停止由{local_ip} 批量 Ping {net_prefix}{start} - {net_prefix}{end}") - self.ping_fun.stop_batch_ping() - self.batchIP_startPing['btn'].config(state='normal') + def on_task_done(self): + self.after(0, lambda: self._set_start_buttons("normal")) - def batchIP_ping_callback(self): - self.batchIP_startPing['btn'].config(state='normal') - - - - - - - - \ No newline at end of file + def _set_start_buttons(self, state): + self.start_btn.configure(state=state) + self.batch_start_btn.configure(state=state) diff --git a/core/ui/tab_telnet.py b/core/ui/tab_telnet.py index 564b654..03a9365 100644 --- a/core/ui/tab_telnet.py +++ b/core/ui/tab_telnet.py @@ -1,135 +1,328 @@ import tkinter as tk -from tkinter import ttk, filedialog, messagebox, scrolledtext +from tkinter import filedialog, messagebox, ttk -from core.ui.basic_ui import BasicUI from core.Function.telnet_fun import PortScanner +from core.ui.components import Console, Page, action_bar, button, combo, field -import logging -logger = logging.getLogger(__name__) -class TelnetTab(ttk.Frame, BasicUI): +PORT_PRESETS = { + "常用端口": "21,22,23,25,53,80,110,139,143,443,445,1433,3306,3389,5432,6379,8080,8443,9200", + "Web 服务": "80,443,8000-8010,8080,8081,8443,8888,9000", + "数据库": "1433,1521,3306,5432,6379,9200,27017", + "远程管理": "22,23,3389,5900,5985,5986", + "邮件服务": "25,110,143,465,587,993,995", + "全端口": "1-65535", +} + + +class TelnetTab(Page): def __init__(self, parent): - super().__init__(parent) - self.telnet_ui() - self.telnet_fun = PortScanner(self.result_box) + super().__init__(parent, "端口扫描", "单端口测试、端口扫描、批量主机巡检、服务识别与结果导出。") + self.body.rowconfigure(3, weight=1) - def telnet_ui(self): - """端口扫描界面布局""" - self.create_assigntelnet_section() - self.create_listtelnet_section() - self.create_batchtelnet_section() - self.create_outputping_section() -# --------------------------------------UI界面布局函数-------------------------------------- - def create_assigntelnet_section(self): - # 区域标签 - frame = ttk.LabelFrame(self, text="指定端口连接测试") - frame.pack(side='top', fill='x', padx=10, pady=5) + status = self.section("实时状态", 0, columns=7) + self.state = field(status, "状态", 1, 0, "等待", 10) + self.total = field(status, "总数", 1, 1, "0", 8) + self.scanned = field(status, "已扫", 1, 2, "0", 8) + self.open_count = field(status, "开放", 1, 3, "0", 8) + self.closed_count = field(status, "关闭", 1, 4, "0", 8) + self.timeout_count = field(status, "超时", 1, 5, "0", 8) + self.progress = field(status, "进度 / 耗时", 1, 6, "0.0% / 0.0s", 16) + for item in (self.state, self.total, self.scanned, self.open_count, self.closed_count, self.timeout_count, self.progress): + item["entry"].configure(state="disabled") - #self.entry_assignIP_A = self.add_input(frame, "本地IP", row=0, col=0) - self.assignTelnet_IP = self.add_input(frame, "目标IP", row=0, col=0, inivar="202.89.233.100") - self.assignTelnet_port = self.add_input(frame, "目标端口", row=0, col=1, inivar="443", entry_width=10) - self.assignTelnet_test = self.add_button(frame, "测试", row=0, col=2, command=self.btn_assignTelnet_test) + modes = self.section("扫描模式", 1, columns=1) + self.tabs = ttk.Notebook(modes) + self.tabs.grid(row=1, column=0, sticky="ew", padx=18, pady=(0, 18)) + self.single_tab = ttk.Frame(self.tabs, style="Panel.TFrame") + self.scan_tab = ttk.Frame(self.tabs, style="Panel.TFrame") + self.batch_tab = ttk.Frame(self.tabs, style="Panel.TFrame") + self.tabs.add(self.single_tab, text="单端口测试") + self.tabs.add(self.scan_tab, text="端口扫描") + self.tabs.add(self.batch_tab, text="批量主机扫描") + self.build_single_tab() + self.build_scan_tab() + self.build_batch_tab() - def create_listtelnet_section(self): - # 区域标签 - frame = ttk.LabelFrame(self, text="批量端口连接测试") - frame.pack(side='top', fill='x', padx=10, pady=5) + results = self.section("结果列表", 2, columns=1) + results.rowconfigure(1, weight=1) + results.columnconfigure(0, weight=1) + self.results_tree = ttk.Treeview( + results, + columns=("host", "resolved_ip", "port", "service", "status", "latency", "banner"), + show="headings", + height=8, + ) + headings = { + "host": "目标", + "resolved_ip": "解析 IP", + "port": "端口", + "service": "服务", + "status": "状态", + "latency": "延迟", + "banner": "Banner / 错误", + } + widths = { + "host": 150, + "resolved_ip": 130, + "port": 70, + "service": 100, + "status": 80, + "latency": 80, + "banner": 320, + } + for column, title in headings.items(): + self.results_tree.heading(column, text=title) + self.results_tree.column(column, width=widths[column], anchor="w") + self.results_tree.tag_configure("open", foreground="#15803d") + self.results_tree.tag_configure("timeout", foreground="#b7791f") + self.results_tree.tag_configure("unreachable", foreground="#b7791f") + self.results_tree.tag_configure("dns_error", foreground="#dc2626") + self.results_tree.tag_configure("error", foreground="#dc2626") + self.results_tree.grid(row=1, column=0, sticky="nsew", padx=18, pady=(0, 18)) + result_scroll = ttk.Scrollbar(results, orient="vertical", command=self.results_tree.yview) + result_scroll.grid(row=1, column=1, sticky="ns", pady=(0, 18)) + self.results_tree.configure(yscrollcommand=result_scroll.set) - self.listtelnet_IP = self.add_input(frame, "目标IP", row=0, col=0, inivar="202.89.233.100") - self.listtelnet_ports = self.add_input(frame, "端口列表", row=0, col=1, inivar="21,22,23,25,80,110,143,443,1433,3306,3389", entry_width=40) - self.listtelnet_start = self.add_button(frame, "开始", row=0, col=2, command=self.btn_listTelnet_start) - self.listtelnet_stop = self.add_button(frame, "停止", row=0, col=3, command=self.btn_listTelnet_stop) - - def create_batchtelnet_section(self): - # 区域标签 - frame = ttk.LabelFrame(self, text="批量端口连接测试") - frame.pack(side='top', fill='x', padx=10, pady=5) + output = self.section("输出控制台", 3, columns=1) + output.rowconfigure(1, weight=1) + output.columnconfigure(0, weight=1) + self.console = Console(output, height=12) + self.console.grid(row=1, column=0, sticky="nsew", padx=18, pady=(0, 18)) - self.batchTelnet_IP = self.add_input(frame, "目标IP", row=0, col=0, inivar="202.89.233.100") - self.batchTelnet_port_begin = self.add_input(frame, "起始端口", row=0, col=1, inivar="100", entry_width=10) - self.batchTelnet_port_end = self.add_input(frame, "结束端口", row=0, col=2, inivar="1000", entry_width=10) - self.batchTelnet_start = self.add_button(frame, "开始", row=0, col=3, command=self.btn_batchTelnet_start) - self.batchTelnet_stop = self.add_button(frame, "停止", row=0, col=4, command=self.btn_batchTelnet_stop) + self.scanner = PortScanner(self.write, self.on_task_done, self.update_status, self.add_result) - def create_outputping_section(self): - # 区域标签 - frame = ttk.LabelFrame(self, text="端口扫描结果输出") - frame.pack(side='top', fill='x', padx=10, pady=5) + def build_single_tab(self): + for col in range(5): + self.single_tab.columnconfigure(col, weight=1) - self.result_box = scrolledtext.ScrolledText(frame, width=100, height=20) - self.result_box.pack(pady=10) -# --------------------------------------按钮回调函数-------------------------------------- - def btn_assignTelnet_test(self): - self.IP = self.assignTelnet_IP['var'].get() - self.port = self.assignTelnet_port['var'].get() - if not self.IP: - messagebox.showwarning("输入错误", "请输入目标IP或域名！") - return - if not self.port: - messagebox.showwarning("输入错误", "请输入目标端口！") + self.single_host = field(self.single_tab, "目标 IP / 域名", 0, 0, "127.0.0.1", 26) + self.single_port = field(self.single_tab, "目标端口", 0, 1, "443", 12) + self.single_timeout = field(self.single_tab, "超时 ms", 0, 2, "1000", 10) + self.single_banner_var = tk.BooleanVar(value=False) + self._check(self.single_tab, "Banner 探测", self.single_banner_var, 0, 3) + actions = action_bar(self.single_tab, 1, 5) + self.single_btn = button(actions, "测试连接", self.test_single, "Primary.TButton") + self.single_export_btn = button(actions, "导出 CSV", self.export_results, "Secondary.TButton") + + def build_scan_tab(self): + for col in range(6): + self.scan_tab.columnconfigure(col, weight=1) + + self.scan_host = field(self.scan_tab, "目标 IP / 域名", 0, 0, "127.0.0.1", 26) + self.scan_ports = field(self.scan_tab, "端口列表 / 范围", 0, 1, PORT_PRESETS["常用端口"], 44, colspan=2) + self.scan_preset = combo(self.scan_tab, "预设", 0, 3, list(PORT_PRESETS), "常用端口", 16) + self.scan_timeout = field(self.scan_tab, "超时 ms", 0, 4, "800", 10) + self.scan_workers = field(self.scan_tab, "并发数", 0, 5, "128", 10) + self.scan_show_closed_var = tk.BooleanVar(value=False) + self.scan_banner_var = tk.BooleanVar(value=False) + self._check(self.scan_tab, "显示关闭端口", self.scan_show_closed_var, 1, 0) + self._check(self.scan_tab, "Banner 探测", self.scan_banner_var, 1, 1) + self.scan_preset["combobox"].bind("<>", lambda _event: self.apply_preset(self.scan_preset, self.scan_ports)) + + actions = action_bar(self.scan_tab, 2, 6) + self.scan_start_btn = button(actions, "开始扫描", self.scan_ports_action, "Primary.TButton") + self.scan_stop_btn = button(actions, "停止", self.stop_scan, "Danger.TButton") + self.scan_copy_btn = button(actions, "复制开放端口", self.copy_open_ports, "Secondary.TButton") + self.scan_export_btn = button(actions, "导出 CSV", self.export_results, "Secondary.TButton") + + def build_batch_tab(self): + for col in range(6): + self.batch_tab.columnconfigure(col, weight=1) + + self.batch_hosts = field(self.batch_tab, "目标列表 / CIDR / IP 段", 0, 0, "192.168.1.1-254", 38, colspan=2) + self.batch_ports = field(self.batch_tab, "端口列表 / 范围", 0, 2, "22,80,443,3389", 32, colspan=2) + self.batch_preset = combo(self.batch_tab, "预设", 0, 4, list(PORT_PRESETS), "远程管理", 16) + self.batch_timeout = field(self.batch_tab, "超时 ms", 0, 5, "800", 10) + self.batch_workers = field(self.batch_tab, "并发数", 1, 0, "128", 10) + self.batch_show_closed_var = tk.BooleanVar(value=False) + self.batch_banner_var = tk.BooleanVar(value=False) + self._check(self.batch_tab, "显示关闭端口", self.batch_show_closed_var, 1, 1) + self._check(self.batch_tab, "Banner 探测", self.batch_banner_var, 1, 2) + self.batch_preset["combobox"].bind("<>", lambda _event: self.apply_preset(self.batch_preset, self.batch_ports)) + + actions = action_bar(self.batch_tab, 2, 6) + self.batch_start_btn = button(actions, "批量扫描", self.batch_scan_action, "Primary.TButton") + self.batch_stop_btn = button(actions, "停止", self.stop_scan, "Danger.TButton") + self.import_hosts_btn = button(actions, "导入目标", self.import_hosts, "Secondary.TButton") + self.batch_copy_btn = button(actions, "复制开放端口", self.copy_open_ports, "Secondary.TButton") + self.batch_export_btn = button(actions, "导出 CSV", self.export_results, "Secondary.TButton") + + def _check(self, parent, text, variable, row, column): + frame = ttk.Frame(parent, style="Panel.TFrame") + frame.grid(row=row, column=column, sticky="ew", padx=18, pady=(22, 14)) + ttk.Checkbutton(frame, text=text, variable=variable).pack(anchor="w") + + def write(self, text, tag=None): + self.console.write(text, tag) + + def clear(self): + self.console.clear() + for item in self.results_tree.get_children(): + self.results_tree.delete(item) + self.update_status( + { + "state": "等待", + "total": 0, + "scanned": 0, + "open": 0, + "closed": 0, + "timeout": 0, + "progress": 0, + "elapsed": 0, + } + ) + + def test_single(self): + try: + self.clear() + self.single_btn.configure(state="disabled") + self.scanner.test_connect( + self.single_host["var"].get(), + int(self.single_port["var"].get()), + options={ + "timeout_ms": self.single_timeout["var"].get(), + "workers": 1, + "show_closed": True, + "banner_probe": self.single_banner_var.get(), + }, + ) + except Exception as exc: + self.single_btn.configure(state="normal") + messagebox.showwarning("无法测试端口", str(exc)) + + def scan_ports_action(self): + try: + self.clear() + self._set_scan_buttons("disabled") + self.scanner.start_scan(self.scan_host["var"].get(), self.scan_ports["var"].get(), self.scan_options()) + except Exception as exc: + self._set_scan_buttons("normal") + messagebox.showwarning("无法扫描端口", str(exc)) + + def batch_scan_action(self): + try: + self.clear() + self._set_scan_buttons("disabled") + self.scanner.start_batch_scan(self.batch_hosts["var"].get(), self.batch_ports["var"].get(), self.batch_options()) + except Exception as exc: + self._set_scan_buttons("normal") + messagebox.showwarning("无法批量扫描", str(exc)) + + def stop_scan(self): + try: + self.scanner.stop_scan() + except Exception as exc: + messagebox.showinfo("提示", str(exc)) + + def scan_options(self): + return { + "timeout_ms": self.scan_timeout["var"].get(), + "workers": self.scan_workers["var"].get(), + "show_closed": self.scan_show_closed_var.get(), + "banner_probe": self.scan_banner_var.get(), + } + + def batch_options(self): + return { + "timeout_ms": self.batch_timeout["var"].get(), + "workers": self.batch_workers["var"].get(), + "show_closed": self.batch_show_closed_var.get(), + "banner_probe": self.batch_banner_var.get(), + } + + def apply_preset(self, preset_item, ports_item): + ports_item["var"].set(PORT_PRESETS.get(preset_item["var"].get(), PORT_PRESETS["常用端口"])) + + def import_hosts(self): + path = filedialog.askopenfilename( + title="导入扫描目标", + filetypes=[("文本文件", "*.txt *.csv"), ("所有文件", "*.*")], + ) + if not path: return try: - port = int(self.port) - except ValueError: - messagebox.showwarning("输入错误", "端口必须是整数！") - return - if port < 1 or port > 65535: - messagebox.showwarning("输入错误", "起始端口最小为1，结束端口最大为65535！") - return - logger.info(f"测试{self.IP} 的 {port} 端口连接情况") - self.telnet_fun.test_connect(self.IP, port) + with open(path, "r", encoding="utf-8-sig") as file: + items = [] + for line in file: + items.extend(part.strip() for part in line.replace("，", ",").split(",") if part.strip()) + self.batch_hosts["var"].set(",".join(items)) + self.write(f"已导入 {len(items)} 个目标/表达式\n", "success") + except Exception as exc: + messagebox.showwarning("导入失败", str(exc)) - def btn_batchTelnet_start(self): - self.IP = self.batchTelnet_IP['var'].get() - self.port_begin = self.batchTelnet_port_begin['var'].get() - self.port_end = self.batchTelnet_port_end['var'].get() - if not self.IP: - messagebox.showwarning("输入错误", "请输入目标IP或域名！") - return - if not self.port_begin or not self.port_end: - messagebox.showwarning("输入错误", "请输入目标端口！") + def export_results(self): + path = filedialog.asksaveasfilename( + title="导出端口扫描结果", + defaultextension=".csv", + filetypes=[("CSV 文件", "*.csv")], + ) + if not path: return try: - port_begin = int(self.port_begin) - port_end = int(self.port_end) - except ValueError: - messagebox.showwarning("输入错误", "端口必须是整数！") + self.scanner.export_results(path) + self.write(f"已导出结果: {path}\n", "success") + except Exception as exc: + messagebox.showwarning("导出失败", str(exc)) + + def copy_open_ports(self): + text = self.scanner.open_ports_summary() + if not text: + messagebox.showinfo("提示", "没有可复制的开放端口") return - if port_begin < 1 or port_end > 65535: - messagebox.showwarning("输入错误", "起始端口最小为1，结束端口最大为65535！") - return - logger.info(f"开始测试{self.IP} 的 {port_begin} 到 {port_end} 端口连接情况") - self.telnet_fun.start_range_scan(self.IP, port_begin, port_end) - - def btn_batchTelnet_stop(self): - logger.info(f"停止测试{self.IP} 的端口连接情况") - self.telnet_fun.stop_scan() + self.clipboard_clear() + self.clipboard_append(text) + self.write("已复制开放端口汇总到剪贴板\n", "success") - def btn_listTelnet_start(self): - self.IP = self.listtelnet_IP['var'].get() - self.ports = self.listtelnet_ports['var'].get() - - if not self.ports: - messagebox.showwarning("输入错误", "请输入端口号或范围！") - return + def update_status(self, stats): + def apply(): + values = [ + (self.state, stats.get("state", "等待")), + (self.total, str(stats.get("total", 0))), + (self.scanned, str(stats.get("scanned", 0))), + (self.open_count, str(stats.get("open", 0))), + (self.closed_count, str(stats.get("closed", 0))), + (self.timeout_count, str(stats.get("timeout", 0))), + (self.progress, f"{stats.get('progress', 0):.1f}% / {stats.get('elapsed', 0):.1f}s"), + ] + for item, value in values: + item["entry"].configure(state="normal") + item["var"].set(value) + item["entry"].configure(state="disabled") - # 支持格式：例如 "22,80,443,8080" - try: - ports = [int(p.strip()) for p in self.ports.split(',') if p.strip()] - except ValueError: - messagebox.showwarning("输入错误", "端口必须为整数，用逗号分隔") - return + self.after(0, apply) - logger.info(f"开始测试{self.IP} 的 {ports} 端口连接情况") - self.telnet_fun.start_list_scan(self.IP, ports) + def add_result(self, row): + def apply(): + port = row.get("port") or "-" + latency = f"{row.get('latency_ms', 0):.1f} ms" if row.get("latency_ms") else "-" + detail = row.get("banner") or row.get("error") or "" + self.results_tree.insert( + "", + "end", + values=( + row.get("host", ""), + row.get("resolved_ip", ""), + port, + row.get("service", ""), + row.get("status_text", ""), + latency, + detail, + ), + tags=(row.get("status", ""),), + ) - def btn_listTelnet_stop(self): - logger.info(f"停止测试{self.IP} 的端口连接情况") - self.telnet_fun.stop_scan() + self.after(0, apply) + def _set_scan_buttons(self, state): + self.scan_start_btn.configure(state=state) + self.batch_start_btn.configure(state=state) - - - - - \ No newline at end of file + def on_task_done(self): + self.after( + 0, + lambda: ( + self.single_btn.configure(state="normal"), + self._set_scan_buttons("normal"), + ), + ) diff --git a/core/ui/tab_tracert.py b/core/ui/tab_tracert.py index 81be0d4..fac24c5 100644 --- a/core/ui/tab_tracert.py +++ b/core/ui/tab_tracert.py @@ -1,56 +1,50 @@ -import tkinter as tk -from tkinter import ttk, filedialog, messagebox, scrolledtext +from tkinter import messagebox -from core.ui.basic_ui import BasicUI -from core.Function.tracert_fun import TracertFun +from core.Function.tracert_fun import TracertFun +from core.ui.components import Console, Page, action_bar, button, field -import logging -logger = logging.getLogger(__name__) -class TracertTab(ttk.Frame, BasicUI): +class TracertTab(Page): def __init__(self, parent): - super().__init__(parent) - self.tracert_ui() - self.tracert_fun = TracertFun(self.result_box) + super().__init__(parent, "路由追踪", "查看从本机到目标地址的网络跳点和响应时间。") + self.body.rowconfigure(1, weight=1) - def tracert_ui(self): - """tracert界面布局""" - self.create_targetadd_section() + target = self.section("追踪参数", 0, columns=4) + self.target = field(target, "目标 IP / 域名", 1, 0, "8.8.8.8", 28) + self.max_hops = field(target, "最大跳数", 1, 1, "20", 12) + self.timeout_ms = field(target, "单跳超时（ms）", 1, 2, "800", 14) + actions = action_bar(target, 2, 4) + self.start_btn = button(actions, "↗ 开始追踪", self.start_trace, "Primary.TButton") + self.stop_btn = button(actions, "■ 停止", self.stop_trace, "Danger.TButton") - self.create_output_section() -# --------------------------------------UI界面布局函数-------------------------------------- - def create_targetadd_section(self): - # 区域标签 - frame = ttk.LabelFrame(self, text="追踪目标地址") - frame.pack(side='top', fill='x', padx=10, pady=5) + output = self.section("输出控制台", 1, columns=1) + output.rowconfigure(1, weight=1) + output.columnconfigure(0, weight=1) + self.console = Console(output, height=22) + self.console.grid(row=1, column=0, sticky="nsew", padx=18, pady=(0, 18)) - self.entry_tracert_add = self.add_input(frame, "目标地址", row=0, col=0, entry_width=40, inivar="202.89.233.100") - self.but_tracert_start = self.add_button(frame, "开始追踪", row=0, col=1, width=8, command=self.tracert_start_callback) - self.but_tracert_stop = self.add_button(frame, "停止追踪", row=0, col=2, width=8, command=self.tracert_stop_callback) + self.tracert_fun = TracertFun(self.write, self.on_task_done) - def create_output_section(self): - # 区域标签 - frame = ttk.LabelFrame(self, text="结果输出") - frame.pack(side='top', fill='x', padx=10, pady=5) + def write(self, text, tag=None): + self.console.write(text, tag) - self.result_box = scrolledtext.ScrolledText(frame, width=100, height=25) - self.result_box.pack(pady=10) -# --------------------------------------按钮回调函数-------------------------------------- - def tracert_start_callback(self): - """开始追踪按钮回调""" - target = self.entry_tracert_add['var'].get() - if not target: - messagebox.showwarning("输入错误", "请输入目标地址！") - return - self.tracert_fun.start_tracert(target) + def start_trace(self): + try: + self.console.clear() + self.tracert_fun.start_tracert( + self.target["var"].get(), + int(self.max_hops["var"].get()), + int(self.timeout_ms["var"].get()), + ) + self.start_btn.configure(state="disabled") + except Exception as exc: + messagebox.showwarning("无法开始路由追踪", str(exc)) - def tracert_stop_callback(self): - """停止追踪按钮回调""" - self.tracert_fun.stop_tracert() + def stop_trace(self): + try: + self.tracert_fun.stop_tracert() + except Exception as exc: + messagebox.showinfo("提示", str(exc)) - - - - - - \ No newline at end of file + def on_task_done(self): + self.after(0, lambda: self.start_btn.configure(state="normal")) diff --git a/core/ui/theme.py b/core/ui/theme.py new file mode 100644 index 0000000..5c1b228 --- /dev/null +++ b/core/ui/theme.py @@ -0,0 +1,110 @@ +import tkinter as tk +from tkinter import ttk + + +COLORS = { + "bg": "#eef3f8", + "panel": "#ffffff", + "panel_alt": "#f7f9fc", + "sidebar": "#102033", + "sidebar_hover": "#1a3552", + "sidebar_active": "#246bfe", + "text": "#172033", + "muted": "#607086", + "border": "#dbe4ef", + "primary": "#246bfe", + "primary_dark": "#1451c8", + "danger": "#df3b3b", + "success": "#1a9b6c", + "warning": "#b7791f", + "console_bg": "#0f1724", + "console_fg": "#dbeafe", + "console_muted": "#9fb4d0", +} + + +FONT_BODY = ("Microsoft YaHei UI", 10) +FONT_SMALL = ("Microsoft YaHei UI", 9) +FONT_TITLE = ("Microsoft YaHei UI", 17, "bold") +FONT_SECTION = ("Microsoft YaHei UI", 11, "bold") +FONT_MONO = ("Consolas", 10) + + +def apply_theme(root: tk.Tk) -> None: + root.configure(bg=COLORS["bg"]) + + style = ttk.Style(root) + try: + style.theme_use("clam") + except tk.TclError: + pass + + style.configure(".", font=FONT_BODY) + style.configure("TFrame", background=COLORS["bg"]) + style.configure("Panel.TFrame", background=COLORS["panel"]) + style.configure("Alt.TFrame", background=COLORS["panel_alt"]) + style.configure("TLabel", background=COLORS["panel"], foreground=COLORS["text"]) + style.configure("Muted.TLabel", background=COLORS["panel"], foreground=COLORS["muted"], font=FONT_SMALL) + style.configure("Title.TLabel", background=COLORS["panel"], foreground=COLORS["text"], font=FONT_TITLE) + style.configure("Section.TLabel", background=COLORS["panel"], foreground=COLORS["text"], font=FONT_SECTION) + + style.configure( + "TEntry", + fieldbackground="#ffffff", + foreground=COLORS["text"], + bordercolor=COLORS["border"], + lightcolor=COLORS["border"], + darkcolor=COLORS["border"], + padding=7, + ) + style.configure( + "TCombobox", + fieldbackground="#ffffff", + foreground=COLORS["text"], + bordercolor=COLORS["border"], + arrowcolor=COLORS["muted"], + padding=6, + ) + + style.configure( + "Primary.TButton", + background=COLORS["primary"], + foreground="#ffffff", + borderwidth=0, + focusthickness=0, + padding=(14, 8), + ) + style.map("Primary.TButton", background=[("active", COLORS["primary_dark"]), ("disabled", "#9bb7f5")]) + + style.configure( + "Secondary.TButton", + background="#e7eef8", + foreground=COLORS["text"], + borderwidth=0, + padding=(14, 8), + ) + style.map("Secondary.TButton", background=[("active", "#d6e2f3")]) + + style.configure( + "Danger.TButton", + background=COLORS["danger"], + foreground="#ffffff", + borderwidth=0, + padding=(14, 8), + ) + style.map("Danger.TButton", background=[("active", "#bd2929")]) + + +def build_app_icon(size: int = 64) -> tk.PhotoImage: + image = tk.PhotoImage(width=size, height=size) + image.put(COLORS["sidebar"], to=(0, 0, size, size)) + image.put(COLORS["primary"], to=(8, 8, size - 8, size - 8)) + image.put("#5eead4", to=(16, 20, 27, 31)) + image.put("#ffffff", to=(37, 20, 48, 31)) + image.put("#ffffff", to=(16, 38, 27, 49)) + image.put("#5eead4", to=(37, 38, 48, 49)) + image.put("#ffffff", to=(24, 25, 41, 29)) + image.put("#ffffff", to=(24, 42, 41, 46)) + image.put("#ffffff", to=(21, 28, 25, 42)) + image.put("#ffffff", to=(39, 28, 43, 42)) + return image diff --git a/core/ui/ui_main.py b/core/ui/ui_main.py index eb20deb..46bd097 100644 --- a/core/ui/ui_main.py +++ b/core/ui/ui_main.py @@ -1,31 +1,124 @@ import tkinter as tk -from tkinter import ttk, filedialog, messagebox +from tkinter import ttk -from core.ui.basic_ui import BasicUI +from core.ui.tab_network import NetworkTab from core.ui.tab_ping import PingTab from core.ui.tab_telnet import TelnetTab -from core.ui.tab_network import NetworkTab from core.ui.tab_tracert import TracertTab +from core.ui.theme import COLORS, apply_theme, build_app_icon -import logging -logger = logging.getLogger(__name__) -class MainUI(BasicUI): - def __init__(self, root): +class MainUI: + def __init__(self, root: tk.Tk, base_dir: str = ""): self.root = root - self.root.title("Network tools") # 窗口标题 - self.root.geometry("800x500") # 窗口大小 - root.resizable(False, False) # 禁止水平和垂直调整大小 - # 创建 Notebook 作为多标签页容器 - self.tab_control = ttk.Notebook(root) - self.tab_control.pack(expand=1, fill="both") - # 实例化各个功能页 - self.tabs = { - "网卡设置": NetworkTab(self.tab_control), - "Ping测试": PingTab(self.tab_control), - "端口扫描": TelnetTab(self.tab_control), - "路由追踪": TracertTab(self.tab_control), + self.base_dir = base_dir + self.root.title("NetPilot 网络调试工具") + self.root.geometry("1160x720") + apply_theme(root) + + self.icon_image = build_app_icon() + self.root.iconphoto(True, self.icon_image) + + self.root.columnconfigure(1, weight=1) + self.root.rowconfigure(0, weight=1) + + self.sidebar = tk.Frame(root, width=232, bg=COLORS["sidebar"]) + self.sidebar.grid(row=0, column=0, sticky="ns") + self.sidebar.grid_propagate(False) + + self.content = ttk.Frame(root, style="Panel.TFrame") + self.content.grid(row=0, column=1, sticky="nsew") + self.content.rowconfigure(0, weight=1) + self.content.columnconfigure(0, weight=1) + + self.pages = {} + self.nav_buttons = {} + self._build_sidebar() + self._build_pages() + self.show_page("network") + + def _build_sidebar(self) -> None: + brand = tk.Frame(self.sidebar, bg=COLORS["sidebar"]) + brand.pack(fill="x", padx=20, pady=(24, 26)) + + logo = tk.Label( + brand, + text="NP", + width=4, + height=2, + bg=COLORS["primary"], + fg="#ffffff", + font=("Microsoft YaHei UI", 14, "bold"), + ) + logo.pack(side="left") + + title = tk.Frame(brand, bg=COLORS["sidebar"]) + title.pack(side="left", padx=12) + tk.Label(title, text="NetPilot", bg=COLORS["sidebar"], fg="#ffffff", font=("Microsoft YaHei UI", 16, "bold")).pack(anchor="w") + tk.Label(title, text="网络调试控制台", bg=COLORS["sidebar"], fg="#9fb4d0", font=("Microsoft YaHei UI", 9)).pack(anchor="w") + + items = [ + ("network", "◎", "网卡配置", "IP / DNS / DHCP"), + ("ping", "⌁", "Ping 探测", "单点与批量探活"), + ("ports", "⌕", "端口扫描", "TCP 连通性检测"), + ("trace", "↗", "路由追踪", "跳点路径分析"), + ] + for key, icon, title, subtitle in items: + self.nav_buttons[key] = self._nav_button(key, icon, title, subtitle) + + footer = tk.Label( + self.sidebar, + text="v2 重构版", + bg=COLORS["sidebar"], + fg="#7185a0", + font=("Microsoft YaHei UI", 9), + ) + footer.pack(side="bottom", anchor="w", padx=22, pady=20) + + def _nav_button(self, key: str, icon: str, title: str, subtitle: str) -> tk.Frame: + frame = tk.Frame(self.sidebar, bg=COLORS["sidebar"], cursor="hand2") + frame.pack(fill="x", padx=14, pady=4) + + icon_label = tk.Label(frame, text=icon, width=3, bg=COLORS["sidebar"], fg="#c8d7ec", font=("Microsoft YaHei UI", 18)) + icon_label.pack(side="left", padx=(8, 6), pady=10) + + text_frame = tk.Frame(frame, bg=COLORS["sidebar"]) + text_frame.pack(side="left", fill="x", expand=True) + title_label = tk.Label(text_frame, text=title, bg=COLORS["sidebar"], fg="#ffffff", font=("Microsoft YaHei UI", 10, "bold")) + title_label.pack(anchor="w") + subtitle_label = tk.Label(text_frame, text=subtitle, bg=COLORS["sidebar"], fg="#9fb4d0", font=("Microsoft YaHei UI", 8)) + subtitle_label.pack(anchor="w", pady=(2, 0)) + + widgets = (frame, icon_label, text_frame, title_label, subtitle_label) + for widget in widgets: + widget.bind("", lambda _event, page=key: self.show_page(page)) + widget.bind("", lambda _event, widgets=widgets, page=key: self._paint_nav(widgets, page, hover=True)) + widget.bind("", lambda _event, widgets=widgets, page=key: self._paint_nav(widgets, page, hover=False)) + frame._nav_widgets = widgets + return frame + + def _build_pages(self) -> None: + self.pages = { + "network": NetworkTab(self.content), + "ping": PingTab(self.content), + "ports": TelnetTab(self.content), + "trace": TracertTab(self.content), } - # 添加到 Notebook - for name, tab in self.tabs.items(): - self.tab_control.add(tab, text=name) \ No newline at end of file + for page in self.pages.values(): + page.grid(row=0, column=0, sticky="nsew") + + def show_page(self, key: str) -> None: + self.active_page = key + self.pages[key].tkraise() + for nav_key, frame in self.nav_buttons.items(): + self._paint_nav(frame._nav_widgets, nav_key, hover=False) + + def _paint_nav(self, widgets, key: str, hover: bool) -> None: + active = getattr(self, "active_page", None) == key + bg = COLORS["sidebar_active"] if active else COLORS["sidebar_hover"] if hover else COLORS["sidebar"] + fg = "#ffffff" if active or hover else "#c8d7ec" + muted = "#dce9ff" if active else "#9fb4d0" + for index, widget in enumerate(widgets): + widget.configure(bg=bg) + if isinstance(widget, tk.Label): + widget.configure(fg=fg if index in (1, 3) else muted) diff --git a/main.py b/main.py index 780b197..40c565c 100644 --- a/main.py +++ b/main.py @@ -1,21 +1,25 @@ -import sys import os +import sys import tkinter as tk from core.logger_config import setup_logger from core.ui.ui_main import MainUI -if __name__ == "__main__": - # 进行日志配置 + +def get_base_dir() -> str: + if getattr(sys, "frozen", False): + return os.path.dirname(sys.executable) + return os.path.dirname(os.path.abspath(__file__)) + + +def main() -> None: setup_logger() - # 判断是否是打包后的环境 - if getattr(sys, 'frozen', False): - # 打包后的路径（exe所在的目录） - base_dir = os.path.dirname(sys.executable) - else: - # 普通Python运行时 - base_dir = os.path.dirname(os.path.abspath(__file__)) - # 启动主界面 + root = tk.Tk() - app = MainUI(root) + root.minsize(960, 600) + MainUI(root, base_dir=get_base_dir()) root.mainloop() + + +if __name__ == "__main__": + main() diff --git a/打包.bat b/打包.bat index f93eb63..3a916de 100644 --- a/打包.bat +++ b/打包.bat @@ -1 +1,18 @@ -pyinstaller -F -w main.py -n NetworkTool +@echo off +setlocal + +set "PYTHON_EXE=C:\Users\Administrator\.cache\codex-runtimes\codex-primary-runtime\dependencies\python\python.exe" +set "PYINSTALLER_PATH=%CD%\.packaging\pyinstaller" + +if exist "%PYTHON_EXE%" ( + if not exist "%PYINSTALLER_PATH%\PyInstaller\__main__.py" ( + "%PYTHON_EXE%" -m pip install --target "%PYINSTALLER_PATH%" PyInstaller + if errorlevel 1 exit /b 1 + ) + set "PYTHONPATH=%PYINSTALLER_PATH%" + "%PYTHON_EXE%" -m PyInstaller -F -w main.py -n NetworkTool +) else ( + pyinstaller -F -w main.py -n NetworkTool +) + +endlocal