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

做室内意向图的网站夫唯seo培训

wzjs 2025/7/21 7:54:10
做室内意向图的网站,夫唯seo培训,wordpress换链接,h5模板怎么制作一、项目背景与功能概述 本工具专为华为ONU设备运维设计，支持通过Telnet协议进行设备连接、命令下发与状态监控。核心功能包括： 多预设IP/账号/密码快速连接 实时命令控制台与自动周期指令 系统日志记录与可视化展示 集成Ping网络测试工具 二、代码…

一、项目背景与功能概述

本工具专为华为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())

http://www.dtcms.com/wzjs/34396.html

相关文章：

  • 签订网站建设合同seo查询seo
  • 威海网站建设 孔百度服务电话6988
  • 公司网站开发费用济南兴田德润o评价网页自动点击软件
  • 福建泉州晋江市疫情最新消息seo网络优化教程
  • 做网站文案策划步骤怎么做一个网站页面
  • 郑州制作网站设计建站快车
  • 阜阳做网站哪家好网络公司有哪些
  • 王野摩托车质量可靠吗seo网站推广seo
  • 北京自己怎样做网站免费自助建站
  • 温州建设信息港网站现在什么网络推广好
  • 唐河网站建设100%能上热门的文案
  • 现在哪个网站做网站好百度导航最新版本免费下载
  • 常州市网站制作24小时自助下单平台网站便宜
  • 佛山网站优化公司排名常州网站建设优化
  • 外贸网站建设ppt模板下载seo搜索引擎优化就业前景
  • 网站技术开发想建立自己的网站怎么建立
  • 天津建设网站公司黄页88网站推广方案
  • 做网站的作用南安网站建设
  • 专业模板网站设计公司新媒体推广渠道有哪些
  • 万网做网站如何下载模板网页制作的软件
  • 东南亚cod建站系统网站系统
  • 如何做单位网站东莞搜索排名提升
  • 公司已经有域名 怎么建网站吉安seo网站快速排名
  • 石家庄网站开发公司电话如何做品牌营销
  • 软件dw做网站网站友情链接购买
  • 用代码怎么建设网站网站seo怎么做
  • 百度站内搜索今日新闻热点大事件
  • 河北省建设注册中心网站百度云服务器
  • 天津自动网站建设调试网站优化技巧
  • 网站建设和网络推广外包网页制作平台有哪些