当前位置： 首页 > news >正文

基于Python写的Telnet带GUI客户端

news 来源：原创 2025/5/23 10:22:59

一、项目背景与功能概述

本工具专为华为ONU设备运维设计，支持通过Telnet协议进行设备连接、命令下发与状态监控。核心功能包括：

  • 多预设IP/账号/密码快速连接

  • 实时命令控制台与自动周期指令

  • 系统日志记录与可视化展示

  • 集成Ping网络测试工具

二、代码架构解析

1. GUI框架搭建（PyQt5）
# GUI主窗口定义
class HuaweiONTGUI(QWidget):def __init__(self):super().__init__()self.client = HuaweiONTClient()  # Telnet客户端实例self.init_ui()                    # 初始化界面self.setup_logging()             # 日志系统绑定# 日志处理组件（桥接logging与GUI）
class GuiLogHandler(QObject, logging.Handler):log_signal = pyqtSignal(str)  # 自定义信号用于跨线程通信

关键布局：

  • 采用QHBoxLayout与QVBoxLayout构建左右分栏布局

  • 通过QGroupBox实现功能模块化分组

  • 使用qasync库实现异步事件循环，解决GUI阻塞问题

2. 连接管理模块
# Telnet客户端核心类
class HuaweiONTClient:class LoginState(Enum):  # 登录状态机INIT = auto()WAIT_PASSWORD = auto()COMPLETE = auto()async def connect(self, host, port):self.reader, self.writer = await asyncio.open_connection(host, port)await self._process_welcome_messages()  # 处理欢迎信息async def adaptive_login(self, username, password):return await self._telnet_login(username, password)

技术要点：

  • 状态机管理登录流程（INIT → WAIT_PASSWORD → COMPLETE）

  • 异步处理Telnet选项协商（_process_telnet_options）

  • 智能识别多语言登录提示（支持Login:/Username:等变体）

3. 命令交互系统
# 自动发送功能实现
def toggle_auto_send(self, state):if state == Qt.Checked:interval = int(self.interval_input.text()) * 1000self.auto_send_timer = QTimer()  # 创建定时器self.auto_send_timer.timeout.connect(self.on_auto_send)@asyncSlot()
async def on_auto_send(self):output = await self.client.write(cmd)  # 异步发送指令self.update_cmd_output(output)

交互布置：

  • 支持命令历史记录（QTextEdit持久化显示）

  • 输入框returnPressed事件绑定即时发送

  • 自动命令支持自定义间隔与指令（如定期查询版本信息）

4. 异常处理与日志系统
# 日志信号传递机制
class GuiLogHandler(logging.Handler):def emit(self, record):msg = self.format(record)self.log_signal.emit(msg)  # 信号触发GUI更新# 客户端异常捕获
async def _telnet_login(self, username, password):try:# ...登录流程...except Exception as e:self.logger.error(f"Login error: {str(e)}")return False

可靠性保障：

  • 双日志通道：文件（huawei_ont.log）+ GUI实时显示

  • 异步操作异常捕获与状态回显

  • 网络超时机制（asyncio.TimeoutError处理）

三、项目应用与扩展

典型使用场景：

  • 批量ONU设备固件版本检查

  • 光模块状态实时监控

  • 快速配置下发与故障排查

扩展方向：

  1. 增加SSH协议支持

  2. 实现配置模板化批量部署

  3. 集成SNMP监控模块

  4. 添加拓扑自动发现功能

四、一个成品原码分享

Telnet带GUI客户端原码：

import asyncio
import logging
import re
import sys
import time
from enum import Enum, auto
from PyQt5.QtWidgets import (QApplication, QWidget, QVBoxLayout, QHBoxLayout,QTextEdit, QLineEdit, QPushButton, QLabel, QGroupBox, QFormLayout, QComboBox, QCheckBox)
from PyQt5.QtCore import Qt, QObject, pyqtSignal, QTimer
from PyQt5.QtGui import QFont, QTextOption
from qasync import QEventLoop, asyncSlotclass GuiLogHandler(QObject, logging.Handler):log_signal = pyqtSignal(str)def __init__(self):super().__init__()logging.Handler.__init__(self)self.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))def emit(self, record):msg = self.format(record)self.log_signal.emit(msg)class HuaweiONTGUI(QWidget):def __init__(self):super().__init__()self.client = HuaweiONTClient()self.auto_send_timer = Noneself.init_ui()self.setup_logging()def init_ui(self):self.setWindowTitle('ONT Manager v2.2')self.setGeometry(300, 300, 1000, 700)self.setStyleSheet("""QWidget { background-color: #F5F5F5; }QGroupBox { border: 2px solid #0078D4;border-radius: 5px;margin-top: 1ex;font-weight: bold;}QGroupBox::title { subcontrol-origin: margin; padding: 0 3px; }QTextEdit, QLineEdit { border: 1px solid #CCCCCC;border-radius: 3px;padding: 3px;}QPushButton {background-color: #0078D4;color: white;border: none;padding: 5px 15px;border-radius: 4px;}QPushButton:hover { background-color: #006CBB; }QPushButton:disabled { background-color: #CCCCCC; }""")main_layout = QHBoxLayout()# Left Panelleft_panel = QVBoxLayout()# Connection Infoconn_group = QGroupBox("Connection Settings")form_layout = QFormLayout()self.ip_combo = QComboBox()self.ip_combo.addItems(['192.168.1.1'])self.ip_combo.setEditable(True)self.ip_combo.setCurrentIndex(0)# userself.user_combo = QComboBox()self.user_combo.addItems(['root'])self.user_combo.setEditable(True)# passself.pass_combo = QComboBox()self.pass_combo.addItems(['admin'])self.pass_combo.setEditable(True)self.pass_combo.setInsertPolicy(QComboBox.NoInsert)self.port_input = QLineEdit('23')form_layout.addRow(QLabel('IP:'), self.ip_combo)form_layout.addRow(QLabel('Port:'), self.port_input)form_layout.addRow(QLabel('User:'), self.user_combo)form_layout.addRow(QLabel('Password:'), self.pass_combo)conn_group.setLayout(form_layout)# Buttonsbtn_layout = QHBoxLayout()self.connect_btn = QPushButton('Connect')self.ping_btn = QPushButton('Ping Test')btn_layout.addWidget(self.connect_btn)btn_layout.addWidget(self.ping_btn)# Command Consolecmd_group = QGroupBox("Command Console")cmd_layout = QVBoxLayout()auto_send_layout = QHBoxLayout()self.auto_send_check = QCheckBox("Auto Send (sec)")self.interval_input = QLineEdit('5')self.auto_cmd_input = QLineEdit('display version')self.auto_cmd_input.setPlaceholderText("Auto command...")self.interval_input.setFixedWidth(60)self.auto_cmd_input.setFixedWidth(180)auto_send_layout.addWidget(self.auto_send_check)auto_send_layout.addWidget(self.interval_input)auto_send_layout.addWidget(QLabel("Command:"))auto_send_layout.addWidget(self.auto_cmd_input)auto_send_layout.addStretch()self.cmd_input = QLineEdit()self.cmd_input.setPlaceholderText("Enter command...")self.cmd_input.returnPressed.connect(self.on_send_command)self.cmd_output = QTextEdit()self.cmd_output.setReadOnly(True)self.send_btn = QPushButton('Send Command')cmd_layout.addLayout(auto_send_layout)cmd_layout.addWidget(self.cmd_input)cmd_layout.addWidget(self.send_btn)cmd_layout.addWidget(self.cmd_output)cmd_group.setLayout(cmd_layout)left_panel.addWidget(conn_group)left_panel.addLayout(btn_layout)left_panel.addWidget(cmd_group)# Right Panel (Logs)right_panel = QVBoxLayout()log_group = QGroupBox("System Logs")log_layout = QVBoxLayout()self.log_display = QTextEdit()self.log_display.setReadOnly(True)self.log_display.setFont(QFont("Consolas", 9))self.log_display.setWordWrapMode(QTextOption.WrapAnywhere)log_layout.addWidget(self.log_display)log_group.setLayout(log_layout)right_panel.addWidget(log_group)main_layout.addLayout(left_panel, 60)main_layout.addLayout(right_panel, 40)self.setLayout(main_layout)# Signalsself.connect_btn.clicked.connect(self.on_connect)self.ping_btn.clicked.connect(self.on_ping)self.send_btn.clicked.connect(self.on_send_command)self.auto_send_check.stateChanged.connect(self.toggle_auto_send)def toggle_auto_send(self, state):if state == Qt.Checked:interval = int(self.interval_input.text()) * 1000self.auto_send_timer = QTimer()self.auto_send_timer.timeout.connect(self.on_auto_send)self.auto_send_timer.start(interval)else:if self.auto_send_timer:self.auto_send_timer.stop()self.auto_send_timer = None@asyncSlot()async def on_auto_send(self):cmd = self.auto_cmd_input.text().strip()if not cmd:returnself.update_cmd_output(f"> [AUTO] {cmd}")try:output = await self.client.write(cmd)if output:self.update_cmd_output(output)except Exception as e:self.update_cmd_output(f"Auto Command Error: {str(e)}")def setup_logging(self):self.gui_handler = GuiLogHandler()self.gui_handler.log_signal.connect(self.update_log_display)self.client.logger.addHandler(self.gui_handler)def update_log_display(self, text):self.log_display.append(text)self.log_display.verticalScrollBar().setValue(self.log_display.verticalScrollBar().maximum())@asyncSlot()async def on_connect(self):self.connect_btn.setEnabled(False)try:success = await self.client.connect(host=self.ip_combo.currentText(),port=int(self.port_input.text()))if success:login_success = await self.client.adaptive_login(username=self.user_combo.currentText(),password=self.pass_combo.currentText())if login_success:self.update_cmd_output("\nLogin successful! Ready to send commands.")except Exception as e:self.log_display.append(f"Error: {str(e)}")finally:self.connect_btn.setEnabled(True)@asyncSlot()async def on_ping(self):self.ping_btn.setEnabled(False)try:host = self.ip_combo.currentText()proc = await asyncio.create_subprocess_shell(f"ping -n 4 {host}" if sys.platform == 'win32' else f"ping -c 4 {host}",stdout=asyncio.subprocess.PIPE,stderr=asyncio.subprocess.PIPE)stdout, _ = await proc.communicate()output = stdout.decode('gbk' if sys.platform == 'win32' else 'utf-8', errors='replace')self.log_display.append(f"\nPing Results:\n{output}")except Exception as e:self.log_display.append(f"Ping Error: {str(e)}")finally:self.ping_btn.setEnabled(True)def update_cmd_output(self, text):self.cmd_output.append(text)self.cmd_output.verticalScrollBar().setValue(self.cmd_output.verticalScrollBar().maximum())@asyncSlot()async def on_send_command(self):cmd = self.cmd_input.text().strip()if not cmd:returnself.cmd_input.clear()self.update_cmd_output(f"> {cmd}")try:output = await self.client.write(cmd)if output:self.update_cmd_output(output)except Exception as e:self.update_cmd_output(f"Command Error: {str(e)}")class HuaweiONTClient:class LoginState(Enum):INIT = auto()WAIT_USERNAME = auto()WAIT_PASSWORD = auto()COMPLETE = auto()def __init__(self):self.reader = Noneself.writer = Noneself.buffer = b''self.login_state = self.LoginState.INITself.telnet_commands = {b'\xff\xfb\x01': b'\xff\xfd\x01',b'\xff\xfb\x03': b'\xff\xfd\x03',b'\xff\xfb\x18': b'\xff\xfd\x18'}self.logger = logging.getLogger('HuaweiONT')self.logger.setLevel(logging.DEBUG)formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')file_handler = logging.FileHandler('huawei_ont.log', mode='a')file_handler.setFormatter(formatter)console_handler = logging.StreamHandler()console_handler.setFormatter(formatter)self.logger.addHandler(file_handler)self.logger.addHandler(console_handler)async def _process_telnet_options(self):processed = Falsefor cmd, response in self.telnet_commands.items():while cmd in self.buffer:pos = self.buffer.find(cmd)self.writer.write(response)self.buffer = self.buffer[:pos] + self.buffer[pos+len(cmd):]self.logger.debug(f"Responded to Telnet option: {cmd!r}")processed = Trueawait self.writer.drain()return processedasync def _detect_login_prompt(self):patterns = [b'Login:',b'Username:',b'User name:',b'login:',b'\r\nlogin:',b'\r\nUser:']for pattern in patterns:if pattern in self.buffer:pos = self.buffer.find(pattern)self.buffer = self.buffer[pos + len(pattern):]self.logger.debug(f"Detected login prompt: {pattern!r}")return patternif b'\r\n' in self.buffer and len(self.buffer.split(b'\r\n')[-1].strip()) == 0:self.logger.debug("Detected empty line prompt")return b'implicit_prompt'return Noneasync def _process_welcome_messages(self):welcome_patterns = [b'Welcome Visiting Huawei',b'Copyright by Huawei',b'Huawei Home Gateway',b'Huawei Technologies Co',b'\r\n\r\nlogin:',b'\r\n\r\nLogin:']cleaned = Falsefor pattern in welcome_patterns:if pattern in self.buffer:pos = self.buffer.find(pattern)self.buffer = self.buffer[pos + len(pattern):]self.logger.debug(f"Cleaned welcome pattern: {pattern!r}")cleaned = Truereturn cleanedasync def _smart_expect(self, patterns, timeout=10):if not isinstance(patterns, list):patterns = [patterns]compiled_patterns = []for pattern in patterns:if isinstance(pattern, str):pattern = pattern.encode('ascii')compiled_patterns.append(re.compile(pattern) if b'(' in pattern else pattern)end_time = asyncio.get_event_loop().time() + timeoutwhile True:await self._process_telnet_options()for i, pattern in enumerate(compiled_patterns):if isinstance(pattern, re.Pattern):match = pattern.search(self.buffer)if match:self.buffer = self.buffer[match.end():]return ielif pattern in self.buffer:pos = self.buffer.find(pattern)self.buffer = self.buffer[pos + len(pattern):]return iif asyncio.get_event_loop().time() > end_time:raise asyncio.TimeoutError(f"No patterns matched: {patterns}")if not await self._raw_read(timeout=0.5):await asyncio.sleep(0.1)async def _raw_read(self, timeout=1):try:data = await asyncio.wait_for(self.reader.read(4096), timeout=timeout)if data:self.buffer += dataself.logger.debug(f"Received: {data!r}")return bool(data)except asyncio.TimeoutError:return Falseasync def connect(self, host='192.168.1.1', port=23):try:self.reader, self.writer = await asyncio.open_connection(host, port)self.logger.info(f"Connected to {host}:{port}")await self._raw_read(timeout=2)await self._process_welcome_messages()return Trueexcept Exception as e:self.logger.error(f"Connection failed: {str(e)}")return Falseasync def write(self, data, hide=False):if not isinstance(data, bytes):data = data.encode('ascii')if not data.endswith(b'\r\n'):data += b'\r\n'self.writer.write(data)await self.writer.drain()if hide:self.logger.debug("Sent hidden data (password)")else:self.logger.debug(f"Sent: {data!r}")await asyncio.sleep(0.5)await self._raw_read(timeout=2)return self.buffer.decode('ascii', errors='ignore')async def adaptive_login(self, username='root', password='admin'):return await self._telnet_login(username, password)async def _telnet_login(self, username, password):try:self.login_state = self.LoginState.INITstart_time = asyncio.get_event_loop().time()while self.login_state != self.LoginState.COMPLETE:if asyncio.get_event_loop().time() - start_time > 15:raise asyncio.TimeoutError("Login timeout")if self.login_state == self.LoginState.INIT:prompt = await self._detect_login_prompt()if prompt:await self.write(username)self.login_state = self.LoginState.WAIT_PASSWORDself.logger.debug("Username sent, waiting for password...")else:await self._process_welcome_messages()await asyncio.sleep(0.5)elif self.login_state == self.LoginState.WAIT_PASSWORD:prompt = await self._detect_login_prompt()if prompt == b'implicit_prompt':await self.write(password, hide=True)self.login_state = self.LoginState.COMPLETEelif any(p in self.buffer.lower() for p in [b'password:', b'passwd:']):await self.write(password, hide=True)self.login_state = self.LoginState.COMPLETEawait self._raw_read(timeout=0.5)await asyncio.sleep(1)if b'incorrect' in self.buffer.lower():raise PermissionError("Invalid credentials")self.logger.info("Telnet Login successful")return Trueexcept Exception as e:self.logger.error(f"Login error: {str(e)}")return Falseif __name__ == '__main__':app = QApplication(sys.argv)loop = QEventLoop(app)asyncio.set_event_loop(loop)window = HuaweiONTGUI()window.show()with loop:sys.exit(loop.run_forever())

相关文章：

  • 深度学习相比传统机器学习的优势
  • Python中的并发编程
  • 接口自动化测试框架(pytest+allure+aiohttp+ 用例自动生成)
  • 智能制造：基于AI制造企业解决方案架构设计【附全文阅读】
  • 【修改提问代码-筹款】2022-1-29
  • zustand - 状态管理
  • 5G 核心网切换机制全解析：XN、N2 与移动性注册对比
  • 率先实现混合搜索：使用 Elasticsearch 和 Semantic Kernel
  • 释放创意潜力！快速打造你的AI应用：Dify平台介绍
  • 文化基因算法（Memetic Algorithm）详解：原理、实现与应用
  • 【机器学习】集成学习算法及实现过程
  • 【信息系统项目管理师】第15章：项目风险管理 - 55个经典题目及详解
  • 从原理到实践：一文详解残差网络
  • MYSQL order 、group 与row_number详解
  • 操作系统期末复习（三）——内存管理
  • C#在 .NET 9.0 中启用二进制序列化：配置、风险与替代方案
  • TDengine 高可用——三副本
  • C 语言学习笔记（指针3）
  • Qt C++实现马的遍历问题
  • python 打卡DAY27
  • 网站建设比较好/网页设计代码案例
  • wordpress自由定制导航/seo优化是利用规则提高排名
  • 乐清公司做网站/合肥网站推广电话
  • 家里做网站买什么服务器好/做网络推广需要多少钱
  • 网站功能流程图/最新网域查询入口
  • 网站关键词重要吗/网站优化招聘