正文开始
本人在制造业工厂上班,业务部门对现场的好多机台的数据需要在内网上远程查看,索性就研究了一套批量查询的程序。

image.png (95.41 KB, 下载次数: 0)
下载附件
程序界面图
2025-6-9 09:29 上传
路径管理:
通过 PathManagerDialog 类实现对文件路径的管理和编辑。
路径信息存储在 Excel 文件中,并可以通过 JSON 格式进行查看和修改。
配置管理:
通过 ConfigManagerDialog 类实现对系统配置的管理和编辑。
配置信息同样存储在 Excel 文件中,支持动态更新。
文件扫描与监控:
使用 FileScannerThread 类来定期扫描指定目录中的文件数量。
支持根据当前时间和班次(白班或夜班)自动切换日期。
支持读取不同类型的文件夹(NG 和 OK)并统计文件数量。
用户界面更新:
主窗口 MonitorApp 提供了一个图形用户界面,显示监控状态、日期和班次信息。
支持自动刷新监控数据,并在表格中展示每个设备的不同分类文件的数量。
通知功能:
在文件扫描完成后,通过 Webhook URL 发送通知消息到指定的服务端点。
消息内容包括产线、设备名称以及各类文件的数量。
定时任务与倒计时:
使用 QTimer 实现定时更新日期和班次信息的功能。
显示下一次监控的倒计时,并支持手动刷新时间。
路径查询与打开:
提供按钮点击事件,根据当前日期和班次查找并打开对应的文件夹路径。
日志记录:
设置了日志记录功能,将所有操作和错误信息记录到日志文件中。
自适应布局与字体大小调整:
根据屏幕尺寸自动调整字体大小,确保界面在不同分辨率下的良好显示效果。
多标签页支持:
使用 QTabWidget 实现多标签页功能,方便查看不同产线的数据。
[Python] 纯文本查看 复制代码# -*- coding: utf-8 -*-
import os
import sys
import time
import json
import requests
import logging
from PyQt5.QtWidgets import QApplication, QMainWindow, QPushButton, QVBoxLayout, QWidget, QLabel, QHBoxLayout, QAction, \
QSizePolicy, QTextEdit, QDialog, QLineEdit, QMessageBox, QFileDialog, QComboBox, QTableWidget, QTableWidgetItem, \
QScrollArea, QTabWidget, QToolTip
from PyQt5.QtCore import Qt, QThread, pyqtSignal, QUrl, QTimer
from PyQt5.QtGui import QFont, QColor
from PyQt5.QtGui import QDesktopServices # 导入 QDesktopServices
import pandas as pd
from collections import OrderedDict
# 定义Excel文件路径
PATHS_EXCEL_FILE = "paths.xlsx"
# 读取Excel文件并转换为有序字典
def read_paths_from_excel(file_path):
try:
df = pd.read_excel(file_path, sheet_name='Paths')
ordered_dict = OrderedDict()
for _, row in df.iterrows():
line = row['Line']
machine = row['Machine']
shift = row['Shift']
type_ = row['Type']
path_template = row['Path Template']
if line not in ordered_dict:
ordered_dict[line] = OrderedDict()
if machine not in ordered_dict[line]:
ordered_dict[line][machine] = OrderedDict()
if shift not in ordered_dict[line][machine]:
ordered_dict[line][machine][shift] = OrderedDict()
ordered_dict[line][machine][shift][type_] = path_template
return ordered_dict
except Exception as e:
log_message(f"读取Excel文件 {file_path} 时发生错误: {str(e)}")
return OrderedDict()
# 读取配置信息
def read_config_from_excel(file_path):
try:
config_df = pd.read_excel(file_path, sheet_name='Config', index_col='Key')['Value'].to_dict()
return config_df
except Exception as e:
log_message(f"读取Excel文件 {file_path} 的配置信息时发生错误: {str(e)}")
return {}
# 获取目录中文件数量
def get_directory_file_count(directory):
try:
log_message(f"正在检查路径: {directory}")
if directory is None or not os.path.exists(directory):
log_message(f"路径 {directory} 不存在或为空")
return 0
files = os.listdir(directory)
file_count = len(files)
log_message(f"路径 {directory} 中的文件数量: {file_count}")
return file_count
except PermissionError:
log_message(f"权限不足无法访问路径 {directory}")
return 0
except Exception as e:
log_message(f"读取路径 {directory} 时发生错误: {str(e)}")
return 0
# 设置日志记录
def setup_logging():
log_dir = "C:\\ZTL-log"
if not os.path.exists(log_dir):
os.makedirs(log_dir)
log_file = os.path.join(log_dir, "monitor_log.txt")
logging.basicConfig(filename=log_file, level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# 记录日志消息
def log_message(message):
print(message) # 添加打印输出,方便调试
logging.info(message)
class FileScannerThread(QThread):
update_signal = pyqtSignal(str, str, dict)
finished_signal = pyqtSignal(dict)
def __init__(self, lines_machines_path_templates, push_interval_minutes):
super().__init__()
self.lines_machines_path_templates = lines_machines_path_templates
self.push_interval_minutes = push_interval_minutes
self.running = False
def run(self):
while self.running:
self.run_once()
time.sleep(self.push_interval_minutes * 60)
def run_once(self):
current_time = time.localtime()
current_hour = current_time.tm_hour
current_minute = current_time.tm_min
# 根据当前时间判断白班/夜班
if (current_hour >= 8 and current_hour = 20 or current_hour = 0 and current_hour = 0 and current_time.tm_hour = 8 and current_hour = 20 or current_hour = 0 and current_hour time.mktime(current_time):
remaining_seconds = int(next_run_time - time.mktime(current_time))
minutes_remaining = remaining_seconds // 60
seconds_remaining = remaining_seconds % 60
countdown_text = f"下次监控将在 {minutes_remaining} 分 {seconds_remaining} 秒后进行..."
self.countdown_label.setText(countdown_text)
else:
self.countdown_label.setText("正在执行监控...")
else:
self.countdown_label.setText(f"下次监控将在 {self.push_interval_minutes} 分钟后进行...")
def next_run_time(self, current_time):
current_hour = current_time.tm_hour
current_minute = current_time.tm_min
current_second = current_time.tm_sec
minutes_since_last_push = current_minute % self.push_interval_minutes
minutes_to_next_push = self.push_interval_minutes - minutes_since_last_push
next_run_time = time.struct_time((
current_time.tm_year,
current_time.tm_mon,
current_time.tm_mday,
current_time.tm_hour,
current_minute + minutes_to_next_push,
0,
current_time.tm_wday,
current_time.tm_yday,
current_time.tm_isdst
))
return time.mktime(next_run_time)
def open_path_manager(self):
if self.path_manager_dialog is None:
self.path_manager_dialog = PathManagerDialog(self)
self.path_manager_dialog.accepted.connect(self.on_path_manager_accepted)
self.path_manager_dialog.rejected.connect(self.on_path_manager_rejected)
self.path_manager_dialog.show()
def on_path_manager_accepted(self):
self.generate_ui_elements()
self.path_manager_dialog = None
def on_path_manager_rejected(self):
self.path_manager_dialog = None
def open_config_manager(self):
if self.config_manager_dialog is None:
self.config_manager_dialog = ConfigManagerDialog(self)
self.config_manager_dialog.accepted.connect(self.on_config_manager_accepted)
self.config_manager_dialog.rejected.connect(self.on_config_manager_rejected)
self.config_manager_dialog.show()
def on_config_manager_accepted(self):
self.generate_ui_elements()
self.config_manager_dialog = None
def on_config_manager_rejected(self):
self.config_manager_dialog = None
def check_and_open_path(self, line_name, machine_name):
current_time = time.localtime()
current_hour = current_time.tm_hour
if (current_hour >= 8 and current_hour = 20 or current_hour 0:
self.current_tab_index = (self.current_tab_index + 1) % self.tab_widget.count()
self.tab_widget.setCurrentIndex(self.current_tab_index)
def mousePressEvent(self, event):
super().mousePressEvent(event)
self.pause_auto_tab_switch()
def pause_auto_tab_switch(self):
self.auto_tab_switch_timer.stop()
self.resume_timer = QTimer(self)
self.resume_timer.singleShot(5000, self.resume_auto_tab_switch)
def resume_auto_tab_switch(self):
self.auto_tab_switch_timer.start(10000)
if __name__ == '__main__':
app = QApplication(sys.argv)
monitor_app = MonitorApp()
monitor_app.showMaximized()
sys.exit(app.exec_())