今天，领导再次来电，告知之前维修过的自助机又出现了问题，这已是第二次出现此类情况。令人更加烦恼的是，目前尚无人对此问题作出回应。我为此感到忧虑，急需找到解决办法。我打算通过IP地址对自助机网络进行ping测试，若发现异常或已关机，便自动将信息发送至科室微信群，让所有值班人员都能看到。将此想法告知AI后，它竟然为我生成了相应的代码。
自助机状态反馈
今天领导又打来电话，告知自助机又出了问题，这让我感到挺无奈的。上次修理才没过多久，它就又出现了故障。目前只有领导提出了这个问题，其他人还没有反馈。我对此感到有些困惑，不清楚这是否是个别现象，还是机器本身存在潜在的问题。我必须尽快查明问题的根本原因，以免之后再次发生类似情况。
机器经过修理后又出现了故障，这让我不得不重新考虑维修的效果。看来上次的维修可能只是治标不治本，根本问题并未被找出。我开始质疑，这台自助机是否真的需要进行一次全面的检查和彻底的维修。此外，这种故障的反复出现，不仅会妨碍自助机的正常运作，还会给大家带来诸多不便。
网络检测思路
我计划利用IP地址对自助机网络进行ping操作，我得知ping操作能够检查自助机与网络之间的连接是否正常。如果ping操作无法成功，那么很可能是因为网络出现了问题，或者自助机已经停止了运行。这种方法操作起来较为简便，执行起来也不复杂，并且可以实时了解自助机的运行状态。如果这个想法能够实现，我们就能迅速得知自助机的运行情况。
对我来说，这种网络检测手段犹如一根及时的救命稻草。过去，在遇到类似问题时，由于不能及时了解设备状况，维修工作总是滞后，随之而来的是不少麻烦。我期盼借助这次的网络检测，能够有效解决设备故障反馈滞后的难题，防止领导再次来电向我反映情况。
巡检机制设定
我打算建立一套巡检体系，这体系包括人工巡检和定期巡检两种方式。人工巡检允许我在闲暇时亲自去检查自助机的网络状态，这样我能更直接地掌握设备状况。而定期巡检则会按照既定的时间周期自动检测自助机。通过这种方式，无论何时，我都能清楚了解自助机的运行状态。
我计划每周对自助机进行一次手动检查，以确保所有细节都得到关注。同时，我初步安排每三个小时进行一次定时检查，以便在不同时间点掌握自助机的运行状况，并迅速发现任何异常。将这两种巡检方法结合使用，将显著提升检测效率，使我能够更有效地关注自助机的运行情况。
AI代码助力
我向AI表述了我的想法，出乎意料的是，它迅速提供了检测所需的代码。这让我感到非常高兴，因为我原本以为编写这样的代码会非常困难，需要投入很多时间和精力。得益于AI提供的代码，我能够更迅速地实现网络检测的构想。这代码宛如一把钥匙，为我打开了解决问题的通道。
AI代码的问世，让我深刻感受到了科技的威力。它能依据我提供的简略说明，迅速生成恰当的代码。这既为我节省了宝贵的时间，又提升了工作的精确度。我计划马上用这些代码进行试验，检验它们是否能够实现我预想的效果。
系统巡检结果
经过系统持续的监控，多数情况下自助机的网络运行平稳。然而，偶尔也会遇到无法ping通的现象，这表明自助机有时可能自行关闭，或者网络出现了暂时的故障。这些不寻常的状况恰好证实了我的观点是正确的。若非有此检测系统，这些问题或许根本无法在初期就被发现。
观察巡检数据，我发现某些时段异常情况较为频发。以每日中午12点到2点为例，这段时间业务量明显增多，自助设备使用率上升，很可能是由于负载过重导致故障。这一现象让我意识到，在业务高峰时段，必须加强对自助设备状态的监控。
后续改进计划
根据现有的巡检数据，我计划对代码进行一番调整。首先，目标是增强系统对异常状况的识别能力，减少错误警报的发生。此外，我还计划添加新的功能，例如对异常情况进行分类，以便区分是网络问题还是自助设备关机。这样的改进可以让值班人员更准确地把握问题类别，进而更有效地处理问题。
此外，我还会将巡检结果告知上级和同事们，以便他们掌握自助机的实际运行状态。这样一来，当他们在使用过程中遇到问题时，就能给出更精确的描述，从而让我能更有效地处理这些问题。总的来说，我会持续优化这个检测系统，确保自助机故障得到更高效的解决。
大家认为我的网络检测方案能否有效应对自助设备的问题？若大家也曾遭遇过类似的设备故障，又是如何处理的？欢迎在评论区分享你的处理经验和见解。若觉得这篇文章对你有帮助，请不要忘记点赞和转发！
import sys
import subprocess
import sqlite3
import time
import threading
from PyQt5.QtWidgets import (QApplication, QMainWindow, QTableWidget, QTableWidgetItem, QPushButton,
QVBoxLayout, QHBoxLayout, QWidget, QLabel, QLineEdit, QMessageBox,
QDialog, QFormLayout, QComboBox, QGroupBox, QTabWidget, QSplitter,
QHeaderView, QCheckBox, QTextEdit, QInputDialog, QAction, QMenu, QToolBar)
from PyQt5.QtCore import Qt, QTimer, pyqtSignal, QObject, QThread
from PyQt5.QtGui import QColor, QIcon, QFont, QPalette
from wxauto import WeChat
class SignalEmitter(QObject):
"""信号发射器，用于在线程间通信"""
status_updated = pyqtSignal(int, str, str)
ping_started = pyqtSignal()
ping_finished = pyqtSignal()
log_updated = pyqtSignal(str)
class DatabaseManager:
"""数据库管理类，负责与SQLite3数据库交互"""
def __init__(self, db_name="self_service_machines.db"):
self.conn = sqlite3.connect(db_name)
self.cursor = self.conn.cursor()
self.create_tables()
def create_tables(self):
"""创建必要的数据库表"""
# 自助机信息表
self.cursor.execute('''
CREATE TABLE IF NOT EXISTS machines (
id INTEGER PRIMARY KEY AUTOINCREMENT,
machine_id TEXT NOT NULL,
name TEXT NOT NULL,
ip TEXT NOT NULL,
location TEXT,
status TEXT DEFAULT "未知"
)
''')
# 配置信息表
self.cursor.execute('''
CREATE TABLE IF NOT EXISTS config (
id INTEGER PRIMARY KEY AUTOINCREMENT,
wechat_group_name TEXT,
ping_interval INTEGER DEFAULT 60,
auto_ping BOOLEAN DEFAULT 0
)
''')
# 初始化配置表（如果为空）
self.cursor.execute("SELECT COUNT(*) FROM config")
if self.cursor.fetchone()[0] == 0:
self.cursor.execute("INSERT INTO config (wechat_group_name, ping_interval, auto_ping) VALUES (?, ?, ?)",
("运维通知群", 60, 0))
self.conn.commit()
# 日志表
self.cursor.execute('''
CREATE TABLE IF NOT EXISTS logs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
message TEXT NOT NULL
)
''')
def add_machine(self, machine_id, name, ip, location):
"""添加自助机信息"""
try:
self.cursor.execute("INSERT INTO machines (machine_id, name, ip, location) VALUES (?, ?, ?, ?)",
(machine_id, name, ip, location))
self.conn.commit()
return True
except sqlite3.Error as e:
print(f"数据库错误: {e}")
return False
def update_machine(self, machine_id, name, ip, location, db_id=None):
"""更新自助机信息"""
try:
if db_id:
# 通过数据库ID更新
self.cursor.execute("UPDATE machines SET machine_id=?, name=?, ip=?, location=? WHERE id=?",
(machine_id, name, ip, location, db_id))
else:
# 通过机器编号更新
self.cursor.execute("UPDATE machines SET name=?, ip=?, location=? WHERE machine_id=?",
(name, ip, location, machine_id))
self.conn.commit()
return True
except sqlite3.Error as e:
print(f"数据库错误: {e}")
return False
def delete_machine(self, db_id):
"""删除自助机信息"""
try:
self.cursor.execute("DELETE FROM machines WHERE id=?", (db_id,))
self.conn.commit()
return True
except sqlite3.Error as e:
print(f"数据库错误: {e}")
return False
def get_all_machines(self):
"""获取所有自助机信息"""
self.cursor.execute("SELECT id, machine_id, name, ip, location, status FROM machines")
return self.cursor.fetchall()
def update_machine_status(self, machine_id, status):
"""更新自助机状态"""
try:
self.cursor.execute("UPDATE machines SET status=? WHERE machine_id=?", (status, machine_id))
self.conn.commit()
except sqlite3.Error as e:
print(f"数据库错误: {e}")
def get_config(self):
"""获取配置信息"""
self.cursor.execute("SELECT wechat_group_name, ping_interval, auto_ping FROM config LIMIT 1")
return self.cursor.fetchone()
def update_config(self, wechat_group_name, ping_interval, auto_ping):
"""更新配置信息"""
try:
self.cursor.execute("UPDATE config SET wechat_group_name=?, ping_interval=?, auto_ping=? WHERE id=1",
(wechat_group_name, ping_interval, auto_ping))
self.conn.commit()
return True
except sqlite3.Error as e:
print(f"数据库错误: {e}")
return False
def add_log(self, message):
"""添加日志记录"""
try:
self.cursor.execute("INSERT INTO logs (message) VALUES (?)", (message,))
self.conn.commit()
except sqlite3.Error as e:
print(f"数据库错误: {e}")
def get_logs(self, limit=100):
"""获取日志记录"""
self.cursor.execute("SELECT timestamp, message FROM logs ORDER BY timestamp DESC LIMIT ?", (limit,))
return self.cursor.fetchall()
class PingThread(QThread):
"""Ping检测线程"""
def __init__(self, machines, signal_emitter):
super().__init__()
self.machines = machines
self.signal_emitter = signal_emitter
def run(self):
"""线程执行函数"""
self.signal_emitter.ping_started.emit()
for idx, machine in enumerate(self.machines):
db_id, machine_id, name, ip, location, _ = machine
status, response_time = self.ping_host(ip)
self.signal_emitter.status_updated.emit(idx, status, response_time)
# 记录状态变化
if status != "在线":
log_msg = f"警告: 自助机 {machine_id} ({name}) 离线! IP: {ip}"
self.signal_emitter.log_updated.emit(log_msg)
# 尝试发送微信通知
self.send_wechat_notification(machine_id, name, ip, location)
self.signal_emitter.ping_finished.emit()
def ping_host(self, host):
"""Ping指定主机并返回状态和响应时间"""
try:
# 根据操作系统选择合适的ping命令
if sys.platform.startswith('win'):
cmd = ['ping', '-n', '1', '-w', '1000', host]
else:
cmd = ['ping', '-c', '1', '-W', '1', host]
start_time = time.time()
result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=2)
end_time = time.time()
if result.returncode == 0:
response_time = f"{int((end_time - start_time) * 1000)}ms"
return "在线", response_time
else:
return "离线", "超时"
except Exception as e:
return "离线", str(e)
def send_wechat_notification(self, machine_id, name, ip, location):
"""发送微信通知"""
try:
# 延迟导入，避免在没有安装wxauto的环境中报错
from wxauto import WeChat
# 获取配置的微信群名
db_manager = DatabaseManager()
config = db_manager.get_config()
wechat_group_name = config[0]
# 初始化微信
wx = WeChat()
# 查找群聊
wx.ChatWith(wechat_group_name)
# 发送消息
message = f"️ 自助机网络异常通知 ️\n\n机器编号: {machine_id}\n机器名称: {name}\nIP地址: {ip}\n机器位置: {location}\n状态: 网络连接失败\n\n请及时处理！"
wx.SendMsg(message)
log_msg = f"已发送微信通知至 {wechat_group_name}"
self.signal_emitter.log_updated.emit(log_msg)
except Exception as e:
log_msg = f"发送微信通知失败: {str(e)}"
self.signal_emitter.log_updated.emit(log_msg)
class MachineDialog(QDialog):
"""添加/编辑自助机对话框"""
def __init__(self, parent=None, machine_data=None):
super().__init__(parent)
self.setWindowTitle("添加自助机" if machine_data is None else "编辑自助机")
self.setMinimumWidth(400)
self.layout = QFormLayout(self)
self.machine_id_edit = QLineEdit()
self.name_edit = QLineEdit()
self.ip_edit = QLineEdit()
self.location_edit = QLineEdit()
if machine_data:
# 编辑模式
db_id, machine_id, name, ip, location, _ = machine_data
self.machine_id_edit.setText(machine_id)
self.machine_id_edit.setReadOnly(True) # 机器编号不可修改
self.name_edit.setText(name)
self.ip_edit.setText(ip)
self.location_edit.setText(location)
self.db_id = db_id
self.layout.addRow("机器编号:", self.machine_id_edit)
self.layout.addRow("机器名称:", self.name_edit)
self.layout.addRow("IP地址:", self.ip_edit)
self.layout.addRow("安装位置:", self.location_edit)
# 按钮布局
button_layout = QHBoxLayout()
self.save_button = QPushButton("保存")
self.cancel_button = QPushButton("取消")
button_layout.addWidget(self.save_button)
button_layout.addWidget(self.cancel_button)
self.layout.addRow(button_layout)
# 连接信号和槽
self.save_button.clicked.connect(self.accept)
self.cancel_button.clicked.connect(self.reject)
def get_data(self):
"""获取对话框中的数据"""
machine_id = self.machine_id_edit.text()
name = self.name_edit.text()
ip = self.ip_edit.text()
location = self.location_edit.text()
if hasattr(self, 'db_id'):
return self.db_id, machine_id, name, ip, location
else:
return machine_id, name, ip, location
class ConfigDialog(QDialog):
"""配置对话框"""
def __init__(self, parent=None, config=None):
super().__init__(parent)
self.setWindowTitle("系统配置")
self.setMinimumWidth(400)
self.layout = QFormLayout(self)
self.wechat_group_edit = QLineEdit()
self.ping_interval_edit = QLineEdit()
self.auto_ping_checkbox = QCheckBox("启用自动巡查")
if config:
wechat_group, interval, auto_ping = config
self.wechat_group_edit.setText(wechat_group)
self.ping_interval_edit.setText(str(interval))
self.auto_ping_checkbox.setChecked(bool(auto_ping))
self.layout.addRow("微信群名称:", self.wechat_group_edit)
self.layout.addRow("巡查间隔(秒):", self.ping_interval_edit)
self.layout.addRow(self.auto_ping_checkbox)
# 按钮布局
button_layout = QHBoxLayout()
self.save_button = QPushButton("保存")
self.cancel_button = QPushButton("取消")
button_layout.addWidget(self.save_button)
button_layout.addWidget(self.cancel_button)
self.layout.addRow(button_layout)
# 连接信号和槽
self.save_button.clicked.connect(self.accept)
self.cancel_button.clicked.connect(self.reject)
def get_config(self):
"""获取配置数据"""
wechat_group = self.wechat_group_edit.text()
try:
ping_interval = int(self.ping_interval_edit.text())
except ValueError:
ping_interval = 60
auto_ping = self.auto_ping_checkbox.isChecked()
return wechat_group, ping_interval, auto_ping
class LogWidget(QWidget):
"""日志显示控件"""
def __init__(self, parent=None):
super().__init__(parent)
self.initUI()
def initUI(self):
"""初始化UI"""
layout = QVBoxLayout(self)
self.log_text = QTextEdit()
self.log_text.setReadOnly(True)
layout.addWidget(self.log_text)
def update_log(self, message):
"""更新日志"""
timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
log_entry = f"[{timestamp}] {message}\n"
self.log_text.append(log_entry)
# 同时保存到数据库
db_manager = DatabaseManager()
db_manager.add_log(message)
def load_logs(self):
"""从数据库加载日志"""
db_manager = DatabaseManager()
logs = db_manager.get_logs()
self.log_text.clear()
for timestamp, message in logs:
self.log_text.append(f"[{timestamp}] {message}")
class MainWindow(QMainWindow):
"""主窗口类"""
def __init__(self):
super().__init__()
self.initUI()
self.db_manager = DatabaseManager()
self.load_machines()
self.load_config()
self.init_auto_ping()
def initUI(self):
"""初始化UI"""
# 设置窗口标题和大小
self.setWindowTitle("自助机网络监控系统")
self.setMinimumSize(1000, 600)
# 创建中心部件
central_widget = QWidget()
self.setCentralWidget(central_widget)
# 创建主布局
main_layout = QVBoxLayout(central_widget)
# 创建顶部工具栏
self.create_toolbar()
# 创建标签页控件
self.tab_widget = QTabWidget()
# 创建机器管理标签页
self.machine_tab = QWidget()
self.setup_machine_tab()
self.tab_widget.addTab(self.machine_tab, "自助机管理")
# 创建日志标签页
self.log_tab = LogWidget()
self.tab_widget.addTab(self.log_tab, "系统日志")
# 创建配置标签页
self.config_tab = QWidget()
self.setup_config_tab()
self.tab_widget.addTab(self.config_tab, "系统配置")
main_layout.addWidget(self.tab_widget)
# 加载样式表
self.load_stylesheet()
# 初始化信号发射器
self.signal_emitter = SignalEmitter()
self.signal_emitter.status_updated.connect(self.update_machine_status)
self.signal_emitter.log_updated.connect(self.log_tab.update_log)
self.signal_emitter.ping_started.connect(self.on_ping_started)
self.signal_emitter.ping_finished.connect(self.on_ping_finished)
def create_toolbar(self):
"""创建顶部工具栏"""
toolbar = QToolBar("主工具栏")
self.addToolBar(toolbar)
# 添加机器按钮
add_action = QAction("添加机器", self)
add_action.triggered.connect(self.add_machine)
toolbar.addAction(add_action)
# 编辑机器按钮
edit_action = QAction("编辑机器", self)
edit_action.triggered.connect(self.edit_machine)
toolbar.addAction(edit_action)
# 删除机器按钮
delete_action = QAction("删除机器", self)
delete_action.triggered.connect(self.delete_machine)
toolbar.addAction(delete_action)
toolbar.addSeparator()
# 巡查按钮
self.ping_action = QAction("立即巡查", self)
self.ping_action.triggered.connect(self.ping_all_machines)
toolbar.addAction(self.ping_action)
# 自动巡查按钮
self.auto_ping_action = QAction("自动巡查", self)
self.auto_ping_action.setCheckable(True)
self.auto_ping_action.toggled.connect(self.toggle_auto_ping)
toolbar.addAction(self.auto_ping_action)
def setup_machine_tab(self):
"""设置机器管理标签页"""
layout = QVBoxLayout(self.machine_tab)
# 创建搜索框和过滤选项
search_layout = QHBoxLayout()
search_label = QLabel("搜索:")
self.search_edit = QLineEdit()
self.search_edit.setPlaceholderText("输入机器编号、名称或IP...")
self.search_edit.textChanged.connect(self.filter_machines)
status_label = QLabel("状态:")
self.status_filter = QComboBox()
self.status_filter.addItems(["全部", "在线", "离线", "未知"])
self.status_filter.currentTextChanged.connect(self.filter_machines)
search_layout.addWidget(search_label)
search_layout.addWidget(self.search_edit)
search_layout.addWidget(status_label)
search_layout.addWidget(self.status_filter)
search_layout.addStretch()
layout.addLayout(search_layout)
# 创建机器表格
self.machine_table = QTableWidget()
self.machine_table.setColumnCount(6)
self.machine_table.setHorizontalHeaderLabels(["ID", "机器编号", "机器名称", "IP地址", "安装位置", "状态"])
self.machine_table.setColumnHidden(0, True) # 隐藏数据库ID列
self.machine_table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch)
self.machine_table.horizontalHeader().setSectionResizeMode(1, QHeaderView.ResizeToContents)
self.machine_table.horizontalHeader().setSectionResizeMode(5, QHeaderView.ResizeToContents)
self.machine_table.setSelectionBehavior(QTableWidget.SelectRows)
self.machine_table.setEditTriggers(QTableWidget.NoEditTriggers)
layout.addWidget(self.machine_table)
# 创建状态统计
status_layout = QHBoxLayout()
self.total_label = QLabel("总数: 0")
self.online_label = QLabel("在线: 0")
self.offline_label = QLabel("离线: 0")
self.unknown_label = QLabel("未知: 0")
status_layout.addWidget(self.total_label)
status_layout.addWidget(self.online_label)
status_layout.addWidget(self.offline_label)
status_layout.addWidget(self.unknown_label)
status_layout.addStretch()
layout.addLayout(status_layout)
def setup_config_tab(self):
"""设置配置标签页"""
layout = QVBoxLayout(self.config_tab)
# 创建配置表单
config_group = QGroupBox("系统配置")
config_layout = QFormLayout()
self.wechat_group_edit = QLineEdit()
self.ping_interval_edit = QLineEdit()
self.auto_ping_checkbox = QCheckBox("启用自动巡查")
config_layout.addRow("微信群名称:", self.wechat_group_edit)
config_layout.addRow("巡查间隔(秒):", self.ping_interval_edit)
config_layout.addRow(self.auto_ping_checkbox)
config_group.setLayout(config_layout)
layout.addWidget(config_group)
# 创建保存按钮
save_button = QPushButton("保存配置")
save_button.clicked.connect(self.save_config)
layout.addWidget(save_button)
layout.addStretch()
def load_machines(self):
"""从数据库加载机器信息"""
machines = self.db_manager.get_all_machines()
self.machine_table.setRowCount(len(machines))
for row, machine in enumerate(machines):
db_id, machine_id, name, ip, location, status = machine
self.machine_table.setItem(row, 0, QTableWidgetItem(str(db_id)))
self.machine_table.setItem(row, 1, QTableWidgetItem(machine_id))
self.machine_table.setItem(row, 2, QTableWidgetItem(name))
self.machine_table.setItem(row, 3, QTableWidgetItem(ip))
self.machine_table.setItem(row, 4, QTableWidgetItem(location))
status_item = QTableWidgetItem(status)
if status == "在线":
status_item.setBackground(QColor(144, 238, 144)) # 浅绿色
elif status == "离线":
status_item.setBackground(QColor(255, 182, 193)) # 浅红色
else:
status_item.setBackground(QColor(211, 211, 211)) # 浅灰色
self.machine_table.setItem(row, 5, status_item)
# 更新状态统计
self.update_status_stats()
def load_config(self):
"""加载配置信息"""
config = self.db_manager.get_config()
if config:
wechat_group, interval, auto_ping = config
self.wechat_group_edit.setText(wechat_group)
self.ping_interval_edit.setText(str(interval))
self.auto_ping_checkbox.setChecked(bool(auto_ping))
self.auto_ping_action.setChecked(bool(auto_ping))
def save_config(self):
"""保存配置信息"""
wechat_group = self.wechat_group_edit.text()
try:
ping_interval = int(self.ping_interval_edit.text())
except ValueError:
QMessageBox.warning(self, "输入错误", "巡查间隔必须是整数!")
return
auto_ping = self.auto_ping_checkbox.isChecked()
if self.db_manager.update_config(wechat_group, ping_interval, auto_ping):
QMessageBox.information(self, "成功", "配置已保存!")
self.init_auto_ping()
else:
QMessageBox.warning(self, "失败", "保存配置时出错!")
def init_auto_ping(self):
"""初始化自动巡查"""
config = self.db_manager.get_config()
if config and config[2]: # 如果启用了自动巡查
self.auto_ping_timer = QTimer()
self.auto_ping_timer.timeout.connect(self.ping_all_machines)
interval = config[1] * 1000 # 转换为毫秒
self.auto_ping_timer.start(interval)
def toggle_auto_ping(self, checked):
"""切换自动巡查状态"""
if checked:
# 启用自动巡查
try:
interval = int(self.ping_interval_edit.text())
self.auto_ping_timer = QTimer()
self.auto_ping_timer.timeout.connect(self.ping_all_machines)
self.auto_ping_timer.start(interval * 1000) # 转换为毫秒
self.auto_ping_checkbox.setChecked(True)
self.db_manager.update_config(
self.wechat_group_edit.text(),
interval,
True
)
self.log_tab.update_log("已启用自动巡查")
except Exception as e:
self.auto_ping_action.setChecked(False)
QMessageBox.warning(self, "错误", f"无法启用自动巡查: {str(e)}")
else:
# 禁用自动巡查
if hasattr(self, 'auto_ping_timer'):
self.auto_ping_timer.stop()
self.auto_ping_checkbox.setChecked(False)
self.db_manager.update_config(
self.wechat_group_edit.text(),
int(self.ping_interval_edit.text() or 60),
False
)
self.log_tab.update_log("已禁用自动巡查")
def add_machine(self):
"""添加新机器"""
dialog = MachineDialog(self)
if dialog.exec_():
machine_id, name, ip, location = dialog.get_data()
if not machine_id or not name or not ip:
QMessageBox.warning(self, "输入错误", "机器编号、名称和IP地址不能为空!")
return
if self.db_manager.add_machine(machine_id, name, ip, location):
QMessageBox.information(self, "成功", "自助机已添加!")
self.load_machines()
self.log_tab.update_log(f"添加新自助机: {machine_id} ({name})")
else:
QMessageBox.warning(self, "失败", "添加自助机时出错!")
def edit_machine(self):
"""编辑选中的机器"""
selected_rows = self.machine_table.selectionModel().selectedRows()
if not selected_rows:
QMessageBox.warning(self, "选择错误", "请先选择要编辑的自助机!")
return
row = selected_rows[0].row()
db_id = int(self.machine_table.item(row, 0).text())
machine_id = self.machine_table.item(row, 1).text()
name = self.machine_table.item(row, 2).text()
ip = self.machine_table.item(row, 3).text()
location = self.machine_table.item(row, 4).text()
status = self.machine_table.item(row, 5).text()
dialog = MachineDialog(self, (db_id, machine_id, name, ip, location, status))
if dialog.exec_():
db_id, machine_id, name, ip, location = dialog.get_data()
if not machine_id or not name or not ip:
QMessageBox.warning(self, "输入错误", "机器编号、名称和IP地址不能为空!")
return
if self.db_manager.update_machine(machine_id, name, ip, location, db_id):
QMessageBox.information(self, "成功", "自助机已更新!")
self.load_machines()
self.log_tab.update_log(f"更新自助机信息: {machine_id} ({name})")
else:
QMessageBox.warning(self, "失败", "更新自助机时出错!")
def delete_machine(self):
"""删除选中的机器"""
selected_rows = self.machine_table.selectionModel().selectedRows()
if not selected_rows:
QMessageBox.warning(self, "选择错误", "请先选择要删除的自助机!")
return
row = selected_rows[0].row()
db_id = int(self.machine_table.item(row, 0).text())
machine_id = self.machine_table.item(row, 1).text()
name = self.machine_table.item(row, 2).text()
reply = QMessageBox.question(
self, "确认删除",
f"确定要删除自助机 '{machine_id} ({name})' 吗?\n此操作不可撤销!",
QMessageBox.Yes | QMessageBox.No
)
if reply == QMessageBox.Yes:
if self.db_manager.delete_machine(db_id):
QMessageBox.information(self, "成功", "自助机已删除!")
self.load_machines()
self.log_tab.update_log(f"删除自助机: {machine_id} ({name})")
else:
QMessageBox.warning(self, "失败", "删除自助机时出错!")
def ping_all_machines(self):
"""Ping所有机器"""
self.machines = self.db_manager.get_all_machines()
if not self.machines:
QMessageBox.information(self, "提示", "没有可巡查的自助机!")
return
self.ping_thread = PingThread(self.machines, self.signal_emitter)
self.ping_thread.start()
def update_machine_status(self, row, status, response_time):
"""更新机器状态"""
if row < self.machine_table.rowCount():
# 更新表格中的状态
status_item = self.machine_table.item(row, 5)
status_text = f"{status} ({response_time})"
status_item.setText(status_text)
if status == "在线":
status_item.setBackground(QColor(144, 238, 144)) # 浅绿色
elif status == "离线":
status_item.setBackground(QColor(255, 182, 193)) # 浅红色
else:
status_item.setBackground(QColor(211, 211, 211)) # 浅灰色
# 更新数据库中的状态
machine_id = self.machine_table.item(row, 1).text()
self.db_manager.update_machine_status(machine_id, status)
# 更新状态统计
self.update_status_stats()
def update_status_stats(self):
"""更新状态统计信息"""
total = self.machine_table.rowCount()
online = 0
offline = 0
unknown = 0
for row in range(total):
status = self.machine_table.item(row, 5).text()
if "在线" in status:
online += 1
elif "离线" in status:
offline += 1
else:
unknown += 1
self.total_label.setText(f"总数: {total}")
self.online_label.setText(f"在线: {online}")
self.offline_label.setText(f"离线: {offline}")
self.unknown_label.setText(f"未知: {unknown}")
def on_ping_started(self):
"""Ping开始时的处理"""
self.ping_action.setEnabled(False)
self.ping_action.setText("巡查中...")
self.log_tab.update_log("开始巡查所有自助机网络状态")
def on_ping_finished(self):
"""Ping完成时的处理"""
self.ping_action.setEnabled(True)
self.ping_action.setText("立即巡查")
self.log_tab.update_log("巡查完成")
def filter_machines(self):
"""根据搜索文本和状态过滤机器"""
search_text = self.search_edit.text().lower()
status_filter = self.status_filter.currentText()
for row in range(self.machine_table.rowCount()):
show_row = True
# 搜索过滤
if search_text:
match = False
for col in range(1, 5): # 检查机器编号、名称、IP和位置列
item = self.machine_table.item(row, col)
if item and search_text in item.text().lower():
match = True
break
if not match:
show_row = False
# 状态过滤
if status_filter != "全部":
status_item = self.machine_table.item(row, 5)
if status_item and status_filter not in status_item.text():
show_row = False
self.machine_table.setRowHidden(row, not show_row)
def load_stylesheet(self):
"""加载样式表"""
self.setStyleSheet("""
QMainWindow {
background-color: #f5f7fa;
}
QTableWidget {
gridline-color: #e0e0e0;
alternate-background-color: #f9f9f9;
selection-background-color: #d0e1f9;
selection-color: #000000;
}
QTableWidget::item {
padding: 5px;
}
QHeaderView::section {
background-color: #4a6fa5;
color: white;
padding: 5px;
border: 1px solid #6c8ebf;
font-weight: bold;
}
QPushButton {
background-color: #4a6fa5;
color: white;
border: none;
padding: 6px 12px;
border-radius: 4px;
}
QPushButton:hover {
background-color: #6c8ebf;
}
QPushButton:pressed {
background-color: #3b5998;
}
QPushButton:disabled {
background-color: #b0b0b0;
}
QToolBar {
background-color: #4a6fa5;
border: none;
padding: 5px;
}
QToolBar QAction {
color: white;
}
QToolBar QAction:hover {
background-color: #6c8ebf;
}
QTabWidget::pane {
border: 1px solid #e0e0e0;
background-color: white;
border-radius: 4px;
padding: 10px;
}
QTabBar::tab {
background-color: #e0e0e0;
color: #333333;
padding: 6px 12px;
border: 1px solid #c0c0c0;
border-bottom: none;
border-top-left-radius: 4px;
border-top-right-radius: 4px;
margin-right: 2px;
}
QTabBar::tab:selected {
background-color: white;
color: #4a6fa5;
font-weight: bold;
}
QGroupBox {
border: 1px solid #c0c0c0;
border-radius: 4px;
margin-top: 10px;
padding: 10px;
}
QGroupBox::title {
subcontrol-origin: margin;
subcontrol-position: top left;
padding: 0 5px;
background-color: white;
color: #4a6fa5;
font-weight: bold;
}
QLabel {
color: #333333;
}
QLineEdit {
border: 1px solid #c0c0c0;
border-radius: 4px;
padding: 5px;
}
QLineEdit:focus {
border: 1px solid #4a6fa5;
outline: none;
}
QCheckBox {
spacing: 5px;
}
QCheckBox::indicator {
width: 16px;
height: 16px;
}
QCheckBox::indicator:checked {
image: url(data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHZpZXdCb3g9IjAgMCAyNCAyNCI+PHBhdGggZD0iTTE5IDNMNyAxNUM1LjggMTQuMiA0LjUgMTQgMyAxNHYyaDFjMS4xIDAgMi4yLjIgMy4xLjZMMTkgNnYtM3oiIGZpbGw9IiM0QTZGQTVBIi8+PC9zdmc+);
}
QMessageBox {
background-color: white;
}
QMessageBox QLabel {
color: #333333;
}
QMessageBox QPushButton {
background-color: #4a6fa5;
color: white;
border: none;
padding: 6px 12px;
border-radius: 4px;
}
QMessageBox QPushButton:hover {
background-color: #6c8ebf;
}
QMessageBox QPushButton:pressed {
background-color: #3b5998;
}
QTextEdit {
border: 1px solid #c0c0c0;
border-radius: 4px;
padding: 5px;
}
QComboBox {
border: 1px solid #c0c0c0;
border-radius: 4px;
padding: 5px;
background-color: white;
combobox-popup: 0;
}
QComboBox::drop-down {
subcontrol-origin: padding;
subcontrol-position: top right;
width: 20px;
border-left-width: 1px;
border-left-color: #c0c0c0;
border-left-style: solid;
border-top-right-radius: 4px;
border-bottom-right-radius: 4px;
background-color: #e0e0e0;
}
QComboBox::down-arrow {
image: url(data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHZpZXdCb3g9IjAgMCAyNCAyNCI+PHBhdGggZD0iTTE3IDEwTD1sLTUgNS01LTV6IiBmaWxsPSIjMzMzMzMzIi8+PC9zdmc+);
width: 10px;
height: 10px;
}
""")
if __name__ == "__main__":
# 确保中文显示正常
import matplotlib
matplotlib.rcParams["font.family"] = ["SimHei", "WenQuanYi Micro Hei", "Heiti TC"]
app = QApplication(sys.argv)
window = MainWindow()
window.show()
sys.exit(app.exec_())
