12
This commit is contained in:
Binary file not shown.
Binary file not shown.
@@ -53,7 +53,7 @@ class PingFun:
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT,
|
||||
universal_newlines=True,
|
||||
creationflags=creationflags # ✅ 加上这个!
|
||||
creationflags=creationflags
|
||||
)
|
||||
for line in iter(self.process.stdout.readline, ''):
|
||||
if self.stop_flag:
|
||||
@@ -127,6 +127,7 @@ class PingFun:
|
||||
return f"{ip} 错误: {e}\n"
|
||||
|
||||
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:
|
||||
|
||||
@@ -0,0 +1,84 @@
|
||||
import subprocess
|
||||
import threading
|
||||
import tkinter as tk
|
||||
from tkinter import scrolledtext, messagebox
|
||||
|
||||
|
||||
class TracertFun:
|
||||
def __init__(self, result_box: scrolledtext.ScrolledText):
|
||||
self.result_box = result_box
|
||||
self.process = None
|
||||
self.stop_flag = False
|
||||
|
||||
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):
|
||||
"""开始追踪"""
|
||||
if self.process:
|
||||
messagebox.showwarning("警告", "⚠️ 正在运行,请先停止再启动。")
|
||||
return
|
||||
|
||||
if not target.strip():
|
||||
messagebox.showwarning("提示", "请输入目标地址!")
|
||||
return
|
||||
|
||||
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 命令"""
|
||||
try:
|
||||
self.process = subprocess.Popen(
|
||||
cmd,
|
||||
shell=True,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT,
|
||||
text=True,
|
||||
creationflags=subprocess.CREATE_NO_WINDOW # 隐藏控制台窗口
|
||||
)
|
||||
|
||||
for line in self.process.stdout:
|
||||
if self.stop_flag:
|
||||
break
|
||||
self._append_text(line)
|
||||
|
||||
except Exception as e:
|
||||
self._append_text(f"\n❌ 错误: {e}\n")
|
||||
|
||||
finally:
|
||||
# 安全关闭进程
|
||||
if self.process:
|
||||
try:
|
||||
self.process.terminate()
|
||||
except Exception:
|
||||
pass
|
||||
self.process = None
|
||||
|
||||
if self.stop_flag:
|
||||
self._append_text("\n=== 已停止追踪 ===\n")
|
||||
else:
|
||||
self._append_text("\n--- 追踪结束 ---\n")
|
||||
|
||||
def stop_tracert(self):
|
||||
"""停止追踪"""
|
||||
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("提示", "当前没有正在运行的追踪任务。")
|
||||
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,56 @@
|
||||
import tkinter as tk
|
||||
from tkinter import ttk, filedialog, messagebox, scrolledtext
|
||||
|
||||
from core.ui.basic_ui import BasicUI
|
||||
from core.Function.tracert_fun import TracertFun
|
||||
|
||||
import logging
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class TracertTab(ttk.Frame, BasicUI):
|
||||
def __init__(self, parent):
|
||||
super().__init__(parent)
|
||||
self.tracert_ui()
|
||||
self.tracert_fun = TracertFun(self.result_box)
|
||||
|
||||
def tracert_ui(self):
|
||||
"""tracert界面布局"""
|
||||
self.create_targetadd_section()
|
||||
|
||||
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)
|
||||
|
||||
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)
|
||||
|
||||
def create_output_section(self):
|
||||
# 区域标签
|
||||
frame = ttk.LabelFrame(self, text="结果输出")
|
||||
frame.pack(side='top', fill='x', padx=10, pady=5)
|
||||
|
||||
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 tracert_stop_callback(self):
|
||||
"""停止追踪按钮回调"""
|
||||
self.tracert_fun.stop_tracert()
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
+3
-1
@@ -5,6 +5,7 @@ from core.ui.basic_ui import BasicUI
|
||||
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
|
||||
|
||||
import logging
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -22,7 +23,8 @@ class MainUI(BasicUI):
|
||||
self.tabs = {
|
||||
"网卡设置": NetworkTab(self.tab_control),
|
||||
"Ping测试": PingTab(self.tab_control),
|
||||
"端口扫描服务端": TelnetTab(self.tab_control),
|
||||
"端口扫描": TelnetTab(self.tab_control),
|
||||
"路由追踪": TracertTab(self.tab_control),
|
||||
}
|
||||
# 添加到 Notebook
|
||||
for name, tab in self.tabs.items():
|
||||
|
||||
Reference in New Issue
Block a user