qsc20001102/Network-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

OK

Browse Source
This commit is contained in:
qsc
2026-06-06 12:34:35 +08:00
parent 00dd907f65
commit 2d67146487
10 changed files with 1682 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/ui/basic_ui.py
-71
View File
@@ -1,71 +0,0 @@
import tkinter as tk
from tkinter import ttk
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",
):
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=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")
return {"frame": group_frame, "label": label, "entry": entry, "var": var}
def add_combobox(
self,
parent,
label_text,
row,
col=0,
listbox=None,
inivar=0,
width=17,
colspan=1,
sticky="w",
label_width=8,
state="readonly",
):
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=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=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}
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}
core/ui/tab_device_discovery.py
+233
View File
@@ -0,0 +1,233 @@
import threading
import tkinter as tk
from tkinter import filedialog, messagebox, ttk
from core.Function.device_discovery_fun import ALL_ADAPTERS, DeviceDiscovery
from core.ui.components import Console, Page, action_bar, button, combo, field
class DeviceDiscoveryTab(Page):
def __init__(self, parent):
super().__init__(parent, "设备发现", "发现局域网在线与 ARP 可见设备，整理 IP、MAC、主机名、厂商与来源网卡。")
self.body.rowconfigure(3, weight=1)
status = self.section("实时状态", 0, columns=6)
self.state = field(status, "状态", 1, 0, "等待", 10)
self.current = field(status, "当前网卡", 1, 1, "-", 18)
self.scan_scope = field(status, "扫描范围", 1, 2, "-", 20)
self.progress = field(status, "进度", 1, 3, "0/0", 10)
self.found = field(status, "发现设备", 1, 4, "0", 8)
self.elapsed = field(status, "耗时", 1, 5, "0.0s", 10)
for item in (self.state, self.current, self.scan_scope, self.progress, self.found, self.elapsed):
item["entry"].configure(state="disabled")
params = self.section("发现参数", 1, columns=6)
self.adapter = combo(params, "检测网卡", 1, 0, [ALL_ADAPTERS], ALL_ADAPTERS, 22)
self.scan_range = field(params, "扫描范围", 1, 1, "", 34, colspan=2)
self.workers = field(params, "并发数", 1, 3, "64", 10)
self.timeout = field(params, "超时 ms", 1, 4, "500", 10)
self.max_hosts = field(params, "最大地址数", 1, 5, "254", 10)
actions = action_bar(params, 2, 6)
self.start_btn = button(actions, "开始发现", self.start_discovery, "Primary.TButton")
self.stop_btn = button(actions, "停止", self.stop_discovery, "Danger.TButton")
self.refresh_btn = button(actions, "刷新网卡", self.load_adapters, "Secondary.TButton")
self.auto_range_btn = button(actions, "自动范围", self.fill_default_range, "Secondary.TButton")
self.copy_btn = button(actions, "复制清单", self.copy_inventory, "Secondary.TButton")
self.export_btn = button(actions, "导出 CSV", self.export_results, "Secondary.TButton")
results = self.section("发现结果", 2, columns=1)
results.rowconfigure(1, weight=1)
results.columnconfigure(0, weight=1)
self.results_tree = ttk.Treeview(
results,
columns=("ip", "mac", "hostname", "vendor", "adapter", "latency", "method", "note"),
show="headings",
height=10,
)
headings = {
"ip": "IP",
"mac": "MAC",
"hostname": "主机名",
"vendor": "厂商",
"adapter": "来源网卡",
"latency": "延迟",
"method": "发现方式",
"note": "备注",
}
widths = {
"ip": 130,
"mac": 150,
"hostname": 180,
"vendor": 130,
"adapter": 150,
"latency": 85,
"method": 90,
"note": 130,
}
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("在线", foreground="#15803d")
self.results_tree.tag_configure("ARP 可见", foreground="#b7791f")
self.results_tree.tag_configure("本机", foreground="#246bfe")
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)
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.discovery = DeviceDiscovery(self.write, self.on_task_done, self.update_status, self.add_result)
self.after(350, self.load_adapters)
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": "等待",
"current": "-",
"scan_range": "-",
"total": 0,
"scanned": 0,
"found": 0,
"elapsed": 0,
}
)
def load_adapters(self):
def worker():
try:
values = self.discovery.get_adapter_choices()
default_range = self.discovery.default_scan_range(ALL_ADAPTERS)
except Exception as exc:
values = [ALL_ADAPTERS]
default_range = ""
self.write(f"读取网卡失败: {exc}\n", "warning")
self.after(0, lambda: self.apply_adapters(values, default_range))
threading.Thread(target=worker, daemon=True).start()
def apply_adapters(self, values, default_range):
values = values or [ALL_ADAPTERS]
values = list(dict.fromkeys(values))
self.adapter["combobox"]["values"] = values
if self.adapter["var"].get() not in values:
self.adapter["var"].set(values[1] if len(values) > 1 else values[0])
if not self.scan_range["var"].get() and default_range:
self.scan_range["var"].set(default_range)
def fill_default_range(self):
try:
value = self.discovery.default_scan_range(self.adapter["var"].get())
if not value:
messagebox.showinfo("提示", "未能根据当前网卡生成扫描范围")
return
self.scan_range["var"].set(value)
self.write(f"已生成安全扫描范围: {value}\n", "success")
except Exception as exc:
messagebox.showwarning("生成失败", str(exc))
def start_discovery(self):
try:
self.clear()
self.start_btn.configure(state="disabled")
self.discovery.start_discovery(self.adapter["var"].get(), self.options())
except Exception as exc:
self.start_btn.configure(state="normal")
messagebox.showwarning("无法开始设备发现", str(exc))
def stop_discovery(self):
try:
self.discovery.stop_discovery()
except Exception as exc:
messagebox.showinfo("提示", str(exc))
def options(self):
return {
"scan_range": self.scan_range["var"].get(),
"workers": self.workers["var"].get(),
"timeout_ms": self.timeout["var"].get(),
"max_hosts": self.max_hosts["var"].get(),
}
def export_results(self):
path = filedialog.asksaveasfilename(
title="导出设备发现结果",
defaultextension=".csv",
filetypes=[("CSV 文件", "*.csv")],
)
if not path:
return
try:
self.discovery.export_results(path)
self.write(f"已导出结果: {path}\n", "success")
except Exception as exc:
messagebox.showwarning("导出失败", str(exc))
def copy_inventory(self):
text = self.discovery.copy_summary()
if not text:
messagebox.showinfo("提示", "还没有可复制的设备清单")
return
self.clipboard_clear()
self.clipboard_append(text)
self.write("已复制设备清单到剪贴板\n", "success")
def update_status(self, stats):
def apply():
total = stats.get("total", 0)
scanned = stats.get("scanned", 0)
values = [
(self.state, stats.get("state", "等待")),
(self.current, stats.get("current") or "-"),
(self.scan_scope, stats.get("scan_range") or "-"),
(self.progress, f"{scanned}/{total}" if total else "0/0"),
(self.found, str(stats.get("found", 0))),
(self.elapsed, f"{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")
self.after(0, apply)
def add_result(self, row):
def apply():
method = row.get("method", "")
note = row.get("note", "")
tag = "本机" if "本机" in note else method
latency = row.get("latency_ms", 0)
try:
latency_text = f"{float(latency):.1f} ms" if float(latency) > 0 else ""
except (TypeError, ValueError):
latency_text = ""
self.results_tree.insert(
"",
"end",
values=(
row.get("ip", ""),
row.get("mac", ""),
row.get("hostname", ""),
row.get("vendor", ""),
row.get("adapter", ""),
latency_text,
method,
note,
),
tags=(tag,),
)
self.after(0, apply)
def on_task_done(self):
self.after(0, lambda: self.start_btn.configure(state="normal"))
core/ui/tab_dns.py
+229
View File
@@ -0,0 +1,229 @@
import threading
import tkinter as tk
from tkinter import filedialog, messagebox, ttk
from core.Function.dns_diag_fun import ALL_ADAPTERS, DEFAULT_DOMAINS, DEFAULT_RECORD_TYPES, DnsDiagnostic
from core.ui.components import Console, Page, action_bar, button, combo, field
class DnsTab(Page):
def __init__(self, parent):
super().__init__(parent, "DNS 诊断", "对比本机 DNS 与常用 DNS 的解析结果、耗时和失败原因。")
self.body.rowconfigure(3, weight=1)
status = self.section("实时状态", 0, columns=6)
self.state = field(status, "状态", 1, 0, "等待", 10)
self.current_dns = field(status, "当前 DNS", 1, 1, "-", 24, colspan=2)
self.target_total = field(status, "目标数", 1, 3, "0", 8)
self.progress = field(status, "进度", 1, 4, "0/0", 10)
self.abnormal = field(status, "异常数", 1, 5, "0", 8)
self.elapsed = field(status, "耗时", 2, 0, "0.0s", 10)
for item in (self.state, self.current_dns, self.target_total, self.progress, self.abnormal, self.elapsed):
item["entry"].configure(state="disabled")
params = self.section("诊断参数", 1, columns=6)
self.adapter = combo(params, "检测网卡", 1, 0, [ALL_ADAPTERS], ALL_ADAPTERS, 22)
self.domains = field(params, "目标域名", 1, 1, DEFAULT_DOMAINS, 34, colspan=2)
self.record_types = field(params, "记录类型", 1, 3, DEFAULT_RECORD_TYPES, 18)
self.timeout = field(params, "超时 ms", 1, 4, "2000", 10)
self.repeat_count = field(params, "重复次数", 1, 5, "1", 10)
self.dns_servers = field(params, "DNS 服务器", 2, 0, "", 60, colspan=6)
actions = action_bar(params, 3, 6)
self.start_btn = button(actions, "开始诊断", self.start_diagnosis, "Primary.TButton")
self.stop_btn = button(actions, "停止", self.stop_diagnosis, "Danger.TButton")
self.refresh_btn = button(actions, "刷新网卡", self.load_adapters, "Secondary.TButton")
self.auto_dns_btn = button(actions, "自动 DNS", self.fill_default_dns, "Secondary.TButton")
self.copy_btn = button(actions, "复制摘要", self.copy_summary, "Secondary.TButton")
self.export_btn = button(actions, "导出 CSV", self.export_results, "Secondary.TButton")
results = self.section("诊断结果", 2, columns=1)
results.rowconfigure(1, weight=1)
results.columnconfigure(0, weight=1)
self.results_tree = ttk.Treeview(
results,
columns=("domain", "type", "server", "status", "elapsed", "values", "verdict"),
show="headings",
height=10,
)
headings = {
"domain": "域名",
"type": "类型",
"server": "DNS 服务器",
"status": "状态",
"elapsed": "耗时",
"values": "解析结果",
"verdict": "错误 / 判断",
}
widths = {
"domain": 160,
"type": 70,
"server": 130,
"status": 80,
"elapsed": 85,
"values": 300,
"verdict": 360,
}
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("正常", foreground="#15803d")
self.results_tree.tag_configure("无记录", foreground="#b7791f")
self.results_tree.tag_configure("失败", 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)
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.diagnostic = DnsDiagnostic(self.write, self.on_task_done, self.update_status, self.add_result)
self.after(350, self.load_adapters)
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": "等待",
"current_dns": "-",
"total": 0,
"completed": 0,
"abnormal": 0,
"elapsed": 0,
}
)
def load_adapters(self):
def worker():
try:
values = self.diagnostic.get_adapter_choices()
default_dns = self.diagnostic.default_dns_servers(values[1] if len(values) > 1 else ALL_ADAPTERS)
except Exception as exc:
values = [ALL_ADAPTERS]
default_dns = ""
self.write(f"读取网卡 DNS 失败: {exc}\n", "warning")
self.after(0, lambda: self.apply_adapters(values, default_dns))
threading.Thread(target=worker, daemon=True).start()
def apply_adapters(self, values, default_dns):
values = values or [ALL_ADAPTERS]
values = list(dict.fromkeys(values))
self.adapter["combobox"]["values"] = values
if self.adapter["var"].get() not in values:
self.adapter["var"].set(values[1] if len(values) > 1 else values[0])
if not self.dns_servers["var"].get() and default_dns:
self.dns_servers["var"].set(default_dns)
def fill_default_dns(self):
try:
value = self.diagnostic.default_dns_servers(self.adapter["var"].get())
if not value:
messagebox.showinfo("提示", "未读取到本机 DNS，已保留常用 DNS 对比服务器")
value = "223.5.5.5,114.114.114.114,8.8.8.8"
self.dns_servers["var"].set(value)
self.write(f"已生成 DNS 对比列表: {value}\n", "success")
except Exception as exc:
messagebox.showwarning("生成失败", str(exc))
def start_diagnosis(self):
try:
self.clear()
self.start_btn.configure(state="disabled")
self.diagnostic.start_diagnosis(self.adapter["var"].get(), self.options())
except Exception as exc:
self.start_btn.configure(state="normal")
messagebox.showwarning("无法开始 DNS 诊断", str(exc))
def stop_diagnosis(self):
try:
self.diagnostic.stop_diagnosis()
except Exception as exc:
messagebox.showinfo("提示", str(exc))
def options(self):
return {
"domains": self.domains["var"].get(),
"record_types": self.record_types["var"].get(),
"dns_servers": self.dns_servers["var"].get(),
"timeout_ms": self.timeout["var"].get(),
"repeat_count": self.repeat_count["var"].get(),
}
def export_results(self):
path = filedialog.asksaveasfilename(
title="导出 DNS 诊断结果",
defaultextension=".csv",
filetypes=[("CSV 文件", "*.csv")],
)
if not path:
return
try:
self.diagnostic.export_results(path)
self.write(f"已导出结果: {path}\n", "success")
except Exception as exc:
messagebox.showwarning("导出失败", str(exc))
def copy_summary(self):
text = self.diagnostic.copy_summary()
if not text:
messagebox.showinfo("提示", "还没有可复制的 DNS 诊断摘要")
return
self.clipboard_clear()
self.clipboard_append(text)
self.write("已复制 DNS 诊断摘要到剪贴板\n", "success")
def update_status(self, stats):
def apply():
total = stats.get("total", 0)
completed = stats.get("completed", 0)
values = [
(self.state, stats.get("state", "等待")),
(self.current_dns, stats.get("current_dns") or "-"),
(self.target_total, str(total)),
(self.progress, f"{completed}/{total}" if total else "0/0"),
(self.abnormal, str(stats.get("abnormal", 0))),
(self.elapsed, f"{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")
self.after(0, apply)
def add_result(self, row):
def apply():
elapsed = row.get("elapsed_ms", 0)
try:
elapsed_text = f"{float(elapsed):.1f} ms" if float(elapsed) > 0 else ""
except (TypeError, ValueError):
elapsed_text = ""
detail = row.get("error") or row.get("verdict", "")
self.results_tree.insert(
"",
"end",
values=(
row.get("domain", ""),
row.get("record_type", ""),
row.get("dns_server", ""),
row.get("status", ""),
elapsed_text,
row.get("values", ""),
detail,
),
tags=(row.get("status", ""),),
)
self.after(0, apply)
def on_task_done(self):
self.after(0, lambda: self.start_btn.configure(state="normal"))
core/ui/ui_main.py
+6
View File
@@ -1,6 +1,8 @@
import tkinter as tk
from tkinter import ttk
from core.ui.tab_device_discovery import DeviceDiscoveryTab
from core.ui.tab_dns import DnsTab
from core.ui.tab_ip_conflict import IpConflictTab
from core.ui.tab_loop import LoopTab
from core.ui.tab_network import NetworkTab
@@ -61,6 +63,8 @@ class MainUI:
items = [
("network", "◎", "网卡配置", "IP / DNS / DHCP"),
("dns", "◇", "DNS 诊断", "解析与服务器对比"),
("devices", "▦", "设备发现", "局域网资产"),
("ping", "⌁", "Ping 探测", "单点与批量探活"),
("ports", "⌕", "端口扫描", "TCP 连通性检测"),
("trace", "↗", "路由追踪", "跳点路径分析"),
@@ -104,11 +108,13 @@ class MainUI:
def _build_pages(self) -> None:
self.pages = {
"network": NetworkTab(self.content),
"dns": DnsTab(self.content),
"ping": PingTab(self.content),
"ports": TelnetTab(self.content),
"trace": TracertTab(self.content),
"loop": LoopTab(self.content),
"ip_conflict": IpConflictTab(self.content),
"devices": DeviceDiscoveryTab(self.content),
}
for page in self.pages.values():
page.grid(row=0, column=0, sticky="nsew")